chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
knotless-serve checks oplog values against their CIDs and syncs past a failing repo
crates/knotless/src/base32.rs+25−0
| @@ -24,6 +24,24 @@ pub fn encode(alphabet: &[u8; 32], bytes: &[u8]) -> String { | ||
out | ||
} | ||
| ||
/// Inverts [`encode`]. Returns `None` for a symbol not in `alphabet`, or | ||
/// bits left over that are not zero or make a whole byte. | ||
pub fn decode(alphabet: &[u8; 32], text: &str) -> Option<Vec<u8>> { | ||
let mut out = Vec::with_capacity(text.len() * 5 / 8); | ||
let (mut buffer, mut bits) = (0u32, 0); | ||
for c in text.bytes() { | ||
let symbol = alphabet.iter().position(|&a| a == c)? as u32; | ||
buffer = (buffer << 5) | symbol; | ||
bits += 5; | ||
if bits >= 8 { | ||
bits -= 8; | ||
out.push((buffer >> bits) as u8); | ||
} | ||
} | ||
let clean = bits < 5 && buffer & ((1 << bits) - 1) == 0; | ||
clean.then_some(out) | ||
} | ||
| ||
/// Encodes a u64 as 13 symbols, so encodings sort the way the integers do. | ||
pub fn encode_u64(alphabet: &[u8; 32], value: u64) -> String { | ||
(0..13) | ||
| @@ -59,6 +77,13 @@ mod tests { | ||
assert_eq!(encode(LOWER, b"fo"), "mzxq"); | ||
assert_eq!(encode(LOWER, b"foo"), "mzxw6"); | ||
assert_eq!(encode(LOWER, b"foobar"), "mzxw6ytboi"); | ||
for bytes in [&b""[..], b"f", b"fo", b"foo", b"foob", b"fooba", b"foobar"] { | ||
assert_eq!(decode(LOWER, &encode(LOWER, bytes)).as_deref(), Some(bytes)); | ||
} | ||
// Not of the alphabet, bits left over set, or a byte's worth over. | ||
assert_eq!(decode(LOWER, "m1"), None); | ||
assert_eq!(decode(LOWER, "mz"), None); | ||
assert_eq!(decode(LOWER, "mzx"), None); | ||
} | ||
| ||
#[test] | ||
crates/knotless/src/drisl.rs+80−0
| @@ -41,3 +41,83 @@ fn json(value: &ciborium::Value) -> Option<serde_json::Value> { | ||
_ => return None, | ||
}) | ||
} | ||
| ||
/// A record's JSON, as atproto writes it, in DRISL: the bytes its CID is | ||
/// of. `None` for what no record holds, as a float, or a link or bytes | ||
/// that do not read. | ||
pub fn encode(record: &serde_json::Value) -> Option<Vec<u8>> { | ||
let mut out = Vec::new(); | ||
ciborium::into_writer(&cbor(record)?, &mut out).ok()?; | ||
Some(out) | ||
} | ||
| ||
fn cbor(value: &serde_json::Value) -> Option<ciborium::Value> { | ||
use base64::Engine; | ||
use ciborium::Value as V; | ||
use serde_json::Value as J; | ||
Some(match value { | ||
J::Null => V::Null, | ||
J::Bool(b) => V::Bool(*b), | ||
J::Number(n) => V::Integer(n.as_i64()?.into()), | ||
J::String(s) => V::Text(s.clone()), | ||
J::Array(items) => V::Array(items.iter().map(cbor).collect::<Option<_>>()?), | ||
J::Object(map) => match (map.len(), map.get("$link"), map.get("$bytes")) { | ||
(1, Some(J::String(link)), _) => { | ||
let cid = crate::base32::decode(crate::base32::LOWER, link.strip_prefix('b')?)?; | ||
V::Tag(42, Box::new(V::Bytes([&[0][..], &cid].concat()))) | ||
} | ||
(1, _, Some(J::String(bytes))) => { | ||
let bytes = bytes.trim_end_matches('='); | ||
let engine = base64::engine::general_purpose::STANDARD_NO_PAD; | ||
V::Bytes(engine.decode(bytes).ok()?) | ||
} | ||
_ => { | ||
let mut pairs = map | ||
.iter() | ||
.map(|(k, v)| Some((k, cbor(v)?))) | ||
.collect::<Option<Vec<_>>>()?; | ||
// DRISL's order: shorter keys first, then bytewise. | ||
pairs.sort_by_key(|(k, _)| (k.len(), k.as_bytes())); | ||
let pairs = pairs.into_iter().map(|(k, v)| (V::Text(k.clone()), v)); | ||
V::Map(pairs.collect()) | ||
} | ||
}, | ||
}) | ||
} | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
use super::*; | ||
use serde_json::json; | ||
| ||
#[test] | ||
fn encodes_records_as_they_read() { | ||
let link = "bafyreigbtj4x7ip5legnfznufuopl4sg4knzc2cof6duas4b3q2fy6swua"; | ||
let record = json!({ | ||
"text": "hi", "n": -3, "ok": true, "none": null, "list": [1, "two"], | ||
"blob": {"$type": "blob", "ref": {"$link": link}, "size": 2}, | ||
"raw": {"$bytes": "AQID"}, | ||
}); | ||
let bytes = encode(&record).unwrap(); | ||
assert_eq!(super::record(&bytes), Some(record)); | ||
// Shorter keys first, then bytewise: a, c, bb. | ||
let ordered = encode(&json!({"bb": 1, "c": 2, "a": 3})).unwrap(); | ||
assert_eq!( | ||
ordered, | ||
[0xa3, 0x61, b'a', 3, 0x61, b'c', 2, 0x62, b'b', b'b', 1] | ||
); | ||
// Padded bytes read as unpadded. | ||
assert_eq!( | ||
encode(&json!({"$bytes": "AQI="})), | ||
encode(&json!({"$bytes": "AQI"})) | ||
); | ||
} | ||
| ||
#[test] | ||
fn refuses_what_no_record_holds() { | ||
assert_eq!(encode(&json!({"f": 1.5})), None); | ||
assert_eq!(encode(&json!({"$link": "zQm"})), None); | ||
assert_eq!(encode(&json!({"$link": "b1"})), None); | ||
assert_eq!(encode(&json!({"$bytes": "!"})), None); | ||
} | ||
} | ||
crates/knotless/src/space_sync.rs+114−0
| @@ -251,6 +251,7 @@ pub async fn read_on( | ||
} | ||
let mut ops = Vec::new(); | ||
let (mut cursor, mut last) = (None::<String>, None); | ||
let mut seen = std::collections::HashSet::new(); | ||
while last.is_none() || cursor.is_some() { | ||
// From where the last page left off, after the first. | ||
let from = std::iter::repeat("cursor").zip(cursor.as_deref()); | ||
| @@ -263,6 +264,9 @@ pub async fn read_on( | ||
page.ops.iter().for_each(|op| hash.apply(op)); | ||
ops.extend(page.ops); | ||
cursor = page.cursor; | ||
// A server that hands back a page it gave would be read forever. | ||
// The head, no cursor, ends the read, so is seen once. | ||
ensure!(seen.insert(cursor.clone()), "the oplog repeats a page"); | ||
// The page that reaches the head carries the commit, a full one not. | ||
last = Some((agreed, mac, page.commit)); | ||
} | ||
| @@ -275,6 +279,59 @@ pub async fn read_on( | ||
Ok(ReadOn { ops, commit, hash }) | ||
} | ||
| ||
/// What a record came to by the end of a read on: its value, as its CID | ||
/// says it is, or none once deleted. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Change { | ||
pub collection: String, | ||
pub rkey: String, | ||
pub value: Option<Value>, | ||
} | ||
| ||
impl ReadOn { | ||
/// Each record of a collection `wanted` that the writes read touched, | ||
/// as the last of them left it. An oplog gives a record's value with | ||
/// the last write to it, as it is when the page is read, so each is | ||
/// checked against that write's CID. One with no value, or another, | ||
/// was written again meanwhile, and does not check out: the repo is | ||
/// read whole instead. | ||
pub fn changes(&self, wanted: impl Fn(&str) -> bool) -> Result<Vec<Change>> { | ||
let mut last = std::collections::BTreeMap::new(); | ||
for op in self.ops.iter().filter(|op| wanted(&op.collection)) { | ||
last.insert((op.collection.as_str(), op.rkey.as_str()), op); | ||
} | ||
let mut changes = Vec::with_capacity(last.len()); | ||
for ((collection, rkey), op) in last { | ||
let value = match &op.cid { | ||
None => None, | ||
Some(cid) => { | ||
let value = op | ||
.value | ||
.as_ref() | ||
.context("a record written with no value")?; | ||
Some(checked(value, cid)?) | ||
} | ||
}; | ||
changes.push(Change { | ||
collection: collection.to_string(), | ||
rkey: rkey.to_string(), | ||
value, | ||
}); | ||
} | ||
Ok(changes) | ||
} | ||
} | ||
| ||
/// `value`, if its DRISL encoding is of `want`, the CID a write gave it. | ||
fn checked(value: &Value, want: &str) -> Result<Value> { | ||
let bytes = drisl::encode(value).context("a record that is not DRISL")?; | ||
ensure!( | ||
cid(&bytes) == want, | ||
"a record that does not come to its CID" | ||
); | ||
Ok(value.clone()) | ||
} | ||
| ||
/// A record in a repo read whole. | ||
#[derive(Debug, Clone, PartialEq)] | ||
pub struct Record { | ||
| @@ -439,6 +496,63 @@ mod tests { | ||
assert_ne!(both.digest(), two.digest()); | ||
} | ||
| ||
#[test] | ||
fn changes_are_last_writes_checked() { | ||
let one = serde_json::json!({"text": "one"}); | ||
let two = serde_json::json!({"text": "two"}); | ||
let cid_of = |v: &Value| cid(&drisl::encode(v).unwrap()); | ||
let op = |collection: &str, rkey: &str, cid: Option<String>, value: Option<&Value>| Op { | ||
collection: collection.into(), | ||
rkey: rkey.into(), | ||
cid, | ||
prev: None, | ||
value: value.cloned(), | ||
}; | ||
let read = |ops: Vec<Op>| ReadOn { | ||
ops, | ||
commit: Commit { | ||
ver: 1, | ||
repo: "did:plc:a".into(), | ||
space: "at://s".into(), | ||
rev: "3l".into(), | ||
data: vec![], | ||
}, | ||
hash: SetHash::default(), | ||
}; | ||
let wanted = |c: &str| c == "a.b.c"; | ||
let change = |rkey: &str, value: Option<&Value>| Change { | ||
collection: "a.b.c".into(), | ||
rkey: rkey.into(), | ||
value: value.cloned(), | ||
}; | ||
// Written, written again and deleted: each as it was left, and of | ||
// the collections wanted alone. A value superseded meanwhile is | ||
// given with no value, and is not looked at. | ||
let ops = vec![ | ||
op("a.b.c", "1", Some(cid_of(&one)), None), | ||
op("a.b.c", "1", Some(cid_of(&two)), Some(&two)), | ||
op("a.b.c", "2", Some(cid_of(&one)), Some(&one)), | ||
op("a.b.c", "2", None, None), | ||
op("x.y.z", "1", Some("bafy".into()), None), | ||
]; | ||
let changes = read(ops).changes(wanted).unwrap(); | ||
assert_eq!(changes, [change("1", Some(&two)), change("2", None)]); | ||
| ||
let fails = |ops| read(ops).changes(wanted).err().map(|e| e.to_string()); | ||
let last = op("a.b.c", "1", Some(cid_of(&one)), None); | ||
assert_eq!( | ||
fails(vec![last]).as_deref(), | ||
Some("a record written with no value") | ||
); | ||
let other = op("a.b.c", "1", Some(cid_of(&one)), Some(&two)); | ||
let said = Some("a record that does not come to its CID"); | ||
assert_eq!(fails(vec![other]).as_deref(), said); | ||
let float = serde_json::json!({"n": 1.5}); | ||
let unencodable = op("a.b.c", "1", Some(cid_of(&one)), Some(&float)); | ||
let said = Some("a record that is not DRISL"); | ||
assert_eq!(fails(vec![unencodable]).as_deref(), said); | ||
} | ||
| ||
#[test] | ||
fn cids() { | ||
// The empty map, {}: as atproto's CID for it. | ||
crates/knotless-serve/src/index.rs+11−0
| @@ -183,6 +183,10 @@ pub struct Index { | ||
/// the stamp, so a backfill and the feed cannot both pass it at once | ||
/// ([`Index::put_as_of`]). | ||
revs: Mutex<()>, | ||
/// A lock for each space, held across its sync, so that no two read | ||
/// one repo at once and the slower puts back what it read over what | ||
/// the other did ([`crate::spaces::sync_with`]). | ||
syncing: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>, | ||
} | ||
| ||
/// What backfills made of an account ([`Index::backfilled`]). | ||
| @@ -927,6 +931,7 @@ impl Index { | ||
stamps: Stamps::new(STAMPS_MAX), | ||
kept: Mutex::new(Kept::new(KEPT_ENTRIES)), | ||
revs: Mutex::default(), | ||
syncing: Mutex::default(), | ||
}; | ||
Ok(index) | ||
} | ||
| @@ -3138,6 +3143,12 @@ impl Index { | ||
.collect() | ||
} | ||
| ||
/// The lock held across a sync of the space `uri`. | ||
pub fn syncing(&self, uri: &str) -> Arc<tokio::sync::Mutex<()>> { | ||
let mut locks = self.syncing.lock().unwrap(); | ||
locks.entry(uri.to_string()).or_default().clone() | ||
} | ||
| ||
/// Records that the space `uri` is synced to the space revision `cursor`. | ||
pub fn set_space_cursor(&self, uri: &str, cursor: &str) -> rusqlite::Result<()> { | ||
let db = self.db.lock().unwrap(); | ||
crates/knotless-serve/src/spaces.rs+34−21
| @@ -31,12 +31,12 @@ use knotless::identity; | ||
use knotless::records::nsid; | ||
use knotless::remote::Remote; | ||
use knotless::space::{self, Space}; | ||
use knotless::space_sync::{self, Op, SetHash}; | ||
use knotless::space_sync::{self, SetHash}; | ||
use knotless::url::RepoUrl; | ||
use knotless::xrpc::{Xrpc, XrpcError}; | ||
use serde::Deserialize; | ||
use serde_json::{Value, json}; | ||
use std::collections::{BTreeSet, HashMap, HashSet}; | ||
use std::collections::{BTreeMap, HashMap, HashSet}; | ||
use std::sync::atomic::Ordering; | ||
use std::sync::{Arc, LazyLock, Mutex}; | ||
use std::time::{Duration, Instant}; | ||
| @@ -95,8 +95,9 @@ pub async fn follow(app: Arc<App>, every: Duration) { | ||
// as while it restarts, it is synced again at the next. | ||
let refused = synced.as_ref().err().is_some_and(space::denied); | ||
let gone = synced.as_ref().err().is_some_and(gone); | ||
// What was written there shows on the next look at it. | ||
if synced.as_ref().is_ok_and(|wrote| *wrote) { | ||
// What was written there shows on the next look at it, as does | ||
// what a sync that failed read of its other repos. | ||
if !matches!(synced, Ok(false)) { | ||
app.moved(&uri); | ||
} | ||
crate::telemetry::space_synced(&uri, synced.map(drop)); | ||
| @@ -350,9 +351,15 @@ pub async fn sync(ctx: &Ctx, index: &Index, uri: &str) -> Result<bool> { | ||
sync_with(ctx, index, &joined).await | ||
} | ||
| ||
/// [`sync`], with `joined`, the space as some member reads it. | ||
/// [`sync`], with `joined`, the space as some member reads it. One sync | ||
/// of a space at a time, so that none puts back what it read over what a | ||
/// later one did. A repo that fails keeps none of the rest from being | ||
/// read: it is the error, and the space's cursor stays where it was, so | ||
/// it is listed, and read, again at the next. | ||
pub async fn sync_with(ctx: &Ctx, index: &Index, joined: &Space) -> Result<bool> { | ||
let uri = joined.uri.as_str(); | ||
let lock = index.syncing(uri); | ||
let _held = lock.lock().await; | ||
let owner = identity::resolve(ctx, authority(uri)).await?; | ||
let host = Xrpc::new(ctx.http.clone(), &owner.pds); | ||
let since = index.spaces()?.into_iter().find(|(u, _)| u == uri); | ||
| @@ -360,12 +367,23 @@ pub async fn sync_with(ctx: &Ctx, index: &Index, joined: &Space) -> Result<bool> | ||
let (writers, now) = joined | ||
.writers_since(&host, &owner.did, since.as_deref()) | ||
.await?; | ||
// A repo written to again while the listing was read is listed twice. | ||
let writers: BTreeSet<String> = writers.into_iter().map(|w| w.did).collect(); | ||
// A repo written to again while the listing was read is listed twice: | ||
// the later says where it is now. One read to there already has | ||
// nothing new. | ||
let writers: BTreeMap<String, String> = | ||
writers.into_iter().map(|w| (w.did, w.repo_rev)).collect(); | ||
let fresh = |(did, rev): &(String, String)| { | ||
let kept = index.space_writer(uri, did).ok().flatten(); | ||
kept.is_none_or(|(at, _)| at != *rev) | ||
}; | ||
let writers: Vec<_> = writers.into_iter().filter(fresh).collect(); | ||
let wrote = !writers.is_empty(); | ||
for did in writers { | ||
member(ctx, index, joined, &did).await?; | ||
let mut failed = None; | ||
for (did, _) in writers { | ||
let read = member(ctx, index, joined, &did).await; | ||
failed = failed.or(read.err()); | ||
} | ||
failed.map_or(Ok(()), Err)?; | ||
if let Some(now) = now { | ||
index.set_space_cursor(uri, &now)?; | ||
} | ||
| @@ -412,7 +430,8 @@ fn indexed(collection: &str) -> bool { | ||
} | ||
| ||
/// Reads on from `rev` in `theirs`'s oplog, where its set hash's state was | ||
/// `state`: each write since, checked against the commit they come to. | ||
/// `state`: each write since, checked against the commit they come to, | ||
/// and each record they leave, against the CID its last write gave it. | ||
async fn read_on( | ||
index: &Index, | ||
uri: &str, | ||
| @@ -424,17 +443,11 @@ async fn read_on( | ||
let hash = SetHash::from_state(state)?; | ||
let read = space_sync::read_on(theirs, kem, rev, hash).await?; | ||
let did = theirs.did(); | ||
// A deletion has no CID; a write with no value was written again later. | ||
let last = |op: &&Op| op.cid.is_none() || op.value.is_some(); | ||
for op in read | ||
.ops | ||
.iter() | ||
.filter(|op| indexed(&op.collection)) | ||
.filter(last) | ||
{ | ||
match &op.value { | ||
Some(value) => index.put_from_space(uri, did, &op.collection, &op.rkey, value)?, | ||
None => index.delete_from_space(uri, did, &op.collection, &op.rkey)?, | ||
for change in read.changes(indexed)? { | ||
let (collection, rkey) = (&change.collection, &change.rkey); | ||
match &change.value { | ||
Some(value) => index.put_from_space(uri, did, collection, rkey, value)?, | ||
None => index.delete_from_space(uri, did, collection, rkey)?, | ||
} | ||
} | ||
index.set_space_writer(uri, did, &read.commit.rev, &read.hash.state())?; | ||
docs/design/access.md+11−3
| @@ -389,8 +389,11 @@ checks what it reads: | ||
the revision it was synced at, every page of it. Each write moves the | ||
set hash kept for it, taking out what it replaced and adding what it | ||
wrote, and the hash must come to the commit the last page carries, which | ||
must be vouched for. A write adds or replaces its record; a deletion (an | ||
op with no CID) takes it out. | ||
must be vouched for. Each record the writes touched is taken as the | ||
last of them left it: one written, with the value the oplog gives it, | ||
which must encode, as DRISL, to that write's CID; one deleted (an op | ||
with no CID) taken out. A last write with no value, or another, was | ||
written again as the page was read, and does not check out. | ||
- A repo never synced, or whose writes do not check out (its oplog no | ||
longer reaching back that far, say), is read whole with | ||
`com.atproto.space.getRepo`: a STAR archive, its commit then each | ||
| @@ -399,7 +402,12 @@ checks what it reads: | ||
deleted meanwhile goes too. | ||
| ||
The set hash and revision are kept per repo with the records, and a repo | ||
is moved on only once what was read checks out. The index starts where it | ||
is moved on only once what was read checks out. A repo the listing names | ||
at the revision it was synced to already is not read again. A repo that | ||
cannot be read either way keeps none of the others from being read, but | ||
holds the space's cursor where it was, so it is listed, and read, again | ||
at the next sync. A space is synced by one sync at a time, so one that | ||
read less cannot put it back over one that read more. The index starts where it | ||
left off at the next start, each space synced as the member it was synced | ||
as before. | ||