~ knotless Search/
↑↓ move↵ open⇧↵ all results

chadtmiller.com/knotless
Switch repository
Loading…

git that lives in your Atmosphere account

knotless.dev
1 collaborator

knotless 0.13.0 Latest 2 packages, 1 image

1
10

A backfill reads each account whole from its PDS, checked, and only those that changed

Merged chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/backfill-archive
crates/knotless/src/jetstream.rs+4−0
View file
@@ -24,6 +24,9 @@ pub struct Event {
pub did: Option<String>,
/// Writes to the knotless collections.
pub ops: Vec<Op>,
/// The revision of the commit that made them, when the frame says:
/// what the index weighs them against a backfill by.
pub rev: Option<String>,
/// Whether it says the account's identity changed: its handle, say.
pub identity: bool,
/// What an `#account` event says of the account at its host: `active`,
@@ -248,6 +251,7 @@ pub fn parse(frame: &str) -> Option<Event> {
record,
});
event.did = Some(text(&payload, "did")?);
event.rev = payload["rev"].as_str().map(str::to_string);
}
}
"identity" | "account" | "sync" => {
crates/knotless-serve/src/web/repos.rs+1−1
View file
@@ -30,7 +30,7 @@ pub(super) async fn wrote(app: &Arc<App>, did: &str, ops: &[(&str, &str, &str)])
rkey: rkey.to_string(),
record: None,
};
let indexed = crate::index_op(&app.ctx, index, did, &op).await;
let indexed = crate::index_op(&app.ctx, index, did, &op, None).await;
indexed.map_err(AppError::internal)?;
}
}
crates/knotless-serve/src/archive.rsnew+383−0
View file
@@ -0,0 +1,383 @@
//! An account's repository read whole, as one CAR from its PDS
//! (`com.atproto.sync.getRepo`), and checked before anything in it is
//! believed: its commit signed by the key the account's DID document names,
//! and every block reached from it the one its CID hashes to.
use anyhow::{Context, Result, ensure};
use ciborium::Value as V;
use knotless::crypto::PublicKey;
use knotless::drisl;
use knotless::env::Ctx;
use sha2::{Digest, Sha256};
use std::collections::HashMap;
/// The most of a repository read, in bytes: a larger one is not read.
pub const MAX: usize = 256 << 20;
/// A repository as its latest commit holds it: the commit's revision, and
/// the records kept.
#[derive(Debug)]
pub struct Archive {
pub rev: String,
pub records: Vec<Record>,
}
#[derive(Debug, PartialEq)]
pub struct Record {
pub collection: String,
pub rkey: String,
pub value: serde_json::Value,
}
/// Reads `did`'s repository from `pds`, at most `max` bytes of it, and
/// keeps the records of the collections `keep` names.
pub async fn fetch(
ctx: &Ctx,
did: &str,
pds: &str,
max: usize,
keep: impl Fn(&str) -> bool,
) -> Result<Archive> {
let doc = knotless::identity::document(ctx, did).await?;
let key = knotless::identity::signing_key(&doc)?;
let url = format!(
"{}/xrpc/com.atproto.sync.getRepo",
pds.trim_end_matches('/')
);
let sent = ctx.http.get(url).query(&[("did", did)]).send().await?;
let mut res = sent.error_for_status()?;
let mut car = Vec::new();
while let Some(chunk) = res.chunk().await? {
car.extend_from_slice(&chunk);
ensure!(car.len() <= max, "{did}'s repository is over {max} bytes");
}
open(&car, did, &key, keep)
}
/// The records of `car`, a repository's CAR, once its commit is found to
/// be `did`'s, signed with `key`, and each block on the way to a record
/// kept is found to be the one its CID names.
pub fn open(
car: &[u8],
did: &str,
key: &PublicKey,
keep: impl Fn(&str) -> bool,
) -> Result<Archive> {
let (root, blocks) = blocks(car)?;
let commit: V = decode(&blocks, &root)?;
let field = |name: &str| map_get(&commit, name);
let author = field("did").and_then(V::as_text);
ensure!(author == Some(did), "the commit is not {did}'s");
let rev = field("rev")
.and_then(V::as_text)
.context("the commit has no rev")?;
let sig = field("sig")
.and_then(V::as_bytes)
.context("the commit is unsigned")?;
key.verify(&unsigned(&commit)?, sig)
.with_context(|| format!("{did}'s commit is not signed by its key"))?;
let data = field("data")
.and_then(drisl::link)
.context("the commit names no data")?;
let mut entries = Vec::new();
walk(&blocks, data, &mut entries)?;
let mut records = Vec::new();
for (path, cid) in entries {
let (collection, rkey) = path.split_once('/').context("a path is not a record's")?;
if !keep(collection) {
continue;
}
// One DRISL holds that no record does, as a float, is left out.
let value = drisl::record(block(&blocks, &cid)?);
records.extend(value.map(|value| Record {
collection: collection.to_string(),
rkey: rkey.to_string(),
value,
}));
}
Ok(Archive {
rev: rev.to_string(),
records,
})
}
/// The bytes the commit's signature is over: the commit without it.
fn unsigned(commit: &V) -> Result<Vec<u8>> {
let fields = commit.as_map().context("the commit is not a map")?;
let kept = fields.iter().filter(|(k, _)| k.as_text() != Some("sig"));
let mut out = Vec::new();
ciborium::into_writer(&V::Map(kept.cloned().collect()), &mut out)?;
Ok(out)
}
/// Each key under the MST node at `cid`, in order, with its value's CID.
fn walk(blocks: &Blocks, cid: &[u8], out: &mut Vec<(String, Vec<u8>)>) -> Result<()> {
let node = decode(blocks, cid)?;
if let Some(left) = map_get(&node, "l").and_then(drisl::link) {
walk(blocks, left, out)?;
}
let entries = map_get(&node, "e").and_then(V::as_array);
let mut last: Vec<u8> = Vec::new();
for entry in entries.context("an MST node has no entries")? {
let shared = map_get(entry, "p").and_then(V::as_integer);
let shared = shared.and_then(|p| usize::try_from(p).ok());
let suffix = map_get(entry, "k").and_then(V::as_bytes);
let value = map_get(entry, "v").and_then(drisl::link);
let parts = shared.zip(suffix).zip(value);
let ((shared, suffix), value) = parts.context("an MST entry is malformed")?;
let mut key = last
.get(..shared)
.context("an MST key shares more than it has")?
.to_vec();
key.extend_from_slice(suffix);
out.push((String::from_utf8(key.clone())?, value.to_vec()));
if let Some(right) = map_get(entry, "t").and_then(drisl::link) {
walk(blocks, right, out)?;
}
last = key;
}
Ok(())
}
fn map_get<'a>(value: &'a V, name: &str) -> Option<&'a V> {
let fields = value.as_map()?;
fields
.iter()
.find(|(k, _)| k.as_text() == Some(name))
.map(|(_, v)| v)
}
/// A CAR's blocks, by their CIDs' bytes.
type Blocks<'a> = HashMap<&'a [u8], &'a [u8]>;
/// The root a CAR names, and its blocks.
fn blocks(car: &[u8]) -> Result<(Vec<u8>, Blocks<'_>)> {
let mut rest = car;
let mut sections = Vec::new();
while !rest.is_empty() {
let len = varint(&mut rest).context("a CAR section has no length")?;
let len = usize::try_from(len)?;
ensure!(len <= rest.len(), "a CAR section runs past its end");
sections.push(&rest[..len]);
rest = &rest[len..];
}
let (header, sections) = sections.split_first().context("the CAR is empty")?;
let header: V = ciborium::from_reader(*header)?;
let roots = map_get(&header, "roots").and_then(V::as_array);
let root = roots.and_then(|r| drisl::link(r.first()?));
let root = root.context("the CAR names no root")?.to_vec();
let mut blocks = HashMap::new();
for section in sections {
// A CIDv1 of a SHA-256: version, codec, the hash's code and length.
ensure!(section.len() >= 36, "a CAR block is too short for its CID");
blocks.insert(&section[..36], &section[36..]);
}
Ok((root, blocks))
}
/// The block `cid` names, once it is found to hash to it.
fn block<'a>(blocks: &Blocks<'a>, cid: &[u8]) -> Result<&'a [u8]> {
let data = blocks.get(cid).context("the CAR is missing a block")?;
let sha256 = cid.len() == 36 && cid[0] == 1 && cid[2..4] == [0x12, 0x20];
ensure!(sha256, "a CID is not a SHA-256 CIDv1");
ensure!(
cid[4..] == Sha256::digest(data)[..],
"a block is not the one its CID names"
);
Ok(data)
}
fn decode(blocks: &Blocks, cid: &[u8]) -> Result<V> {
Ok(ciborium::from_reader(block(blocks, cid)?)?)
}
/// An unsigned LEB128 integer off the front of `bytes`.
fn varint(bytes: &mut &[u8]) -> Option<u64> {
let mut value = 0u64;
for (i, &byte) in bytes.iter().enumerate().take(10) {
value |= u64::from(byte & 0x7f) << (7 * i);
if byte & 0x80 == 0 {
*bytes = &bytes[i + 1..];
return Some(value);
}
}
None
}
#[cfg(test)]
mod tests {
use super::*;
use knotless::crypto::P256Key;
fn cbor(value: &V) -> Vec<u8> {
let mut out = Vec::new();
ciborium::into_writer(value, &mut out).unwrap();
out
}
fn cid(data: &[u8]) -> Vec<u8> {
let mut cid = vec![1, 0x71, 0x12, 0x20];
cid.extend_from_slice(&Sha256::digest(data));
cid
}
fn link(cid: &[u8]) -> V {
let mut bytes = vec![0];
bytes.extend_from_slice(cid);
V::Tag(42, Box::new(V::Bytes(bytes)))
}
fn map(pairs: Vec<(&str, V)>) -> V {
V::Map(
pairs
.into_iter()
.map(|(k, v)| (V::Text(k.into()), v))
.collect(),
)
}
fn put(car: &mut Vec<u8>, section: &[u8]) {
let mut len = section.len();
while len >= 0x80 {
car.push((len as u8 & 0x7f) | 0x80);
len >>= 7;
}
car.push(len as u8);
car.extend_from_slice(section);
}
/// A repository of `did`'s holding two `records`, by path, the second
/// in a subtree right of the first, its commit signed by `key` and then
/// changed as `signed` changes it.
fn repo(did: &str, key: &P256Key, records: &[(&str, V)], signed: impl Fn(&mut V)) -> Vec<u8> {
let mut blocks: Vec<Vec<u8>> = Vec::new();
let mut leaf = |value: &V| {
let bytes = cbor(value);
let cid = cid(&bytes);
blocks.push(bytes);
cid
};
let values: Vec<Vec<u8>> = records.iter().map(|(_, v)| leaf(v)).collect();
// The second record in a subtree to the right of the first.
let (first, second) = (records[0].0.as_bytes(), records[1].0.as_bytes());
let right = map(vec![
("l", V::Null),
(
"e",
V::Array(vec![map(vec![
("p", V::Integer(0.into())),
("k", V::Bytes(second.to_vec())),
("v", link(&values[1])),
("t", V::Null),
])]),
),
]);
let right = leaf(&right);
let root = map(vec![
("l", V::Null),
(
"e",
V::Array(vec![map(vec![
("p", V::Integer(0.into())),
("k", V::Bytes(first.to_vec())),
("v", link(&values[0])),
("t", link(&right)),
])]),
),
]);
let data = leaf(&root);
let mut commit = map(vec![
("did", V::Text(did.into())),
("version", V::Integer(3.into())),
("data", link(&data)),
("rev", V::Text("3l".into())),
("prev", V::Null),
]);
let sig = key.sign(&cbor(&commit));
if let V::Map(fields) = &mut commit {
fields.push((V::Text("sig".into()), V::Bytes(sig)));
}
signed(&mut commit);
let commit = leaf(&commit);
let mut car = Vec::new();
put(
&mut car,
&cbor(&map(vec![
("version", V::Integer(1.into())),
("roots", V::Array(vec![link(&commit)])),
])),
);
for block in &blocks {
let mut section = cid(block);
section.extend_from_slice(block);
put(&mut car, &section);
}
car
}
fn records() -> Vec<(&'static str, V)> {
vec![
(
"dev.knotless.repo/a",
map(vec![("name", V::Text("a".into()))]),
),
(
"dev.knotless.star/b",
map(vec![("subject", V::Text("s".into()))]),
),
]
}
#[test]
fn a_signed_repository_opens_to_its_records() {
let key = P256Key::generate();
let car = repo("did:plc:a", &key, &records(), |_| {});
let all = open(&car, "did:plc:a", &key.public(), |_| true).unwrap();
assert_eq!(all.rev, "3l");
let paths: Vec<String> = all
.records
.iter()
.map(|r| format!("{}/{}", r.collection, r.rkey))
.collect();
assert_eq!(paths, ["dev.knotless.repo/a", "dev.knotless.star/b"]);
assert_eq!(all.records[0].value, serde_json::json!({"name": "a"}));
// Only the collections kept.
let repos = open(&car, "did:plc:a", &key.public(), |c| {
c == "dev.knotless.repo"
})
.unwrap();
assert_eq!(repos.records.len(), 1);
}
#[test]
fn what_is_not_the_accounts_is_refused() {
let key = P256Key::generate();
let car = repo("did:plc:a", &key, &records(), |_| {});
// Another account's, or signed by another key.
assert!(open(&car, "did:plc:b", &key.public(), |_| true).is_err());
let other = P256Key::generate().public();
let err = open(&car, "did:plc:a", &other, |_| true).unwrap_err();
assert!(
format!("{err:#}").contains("not signed by its key"),
"{err:#}"
);
// A commit changed after it was signed.
let changed = repo("did:plc:a", &key, &records(), |commit| {
if let V::Map(fields) = commit {
fields[3].1 = V::Text("4z".into());
}
});
assert!(open(&changed, "did:plc:a", &key.public(), |_| true).is_err());
// A block changed in the CAR: it no longer hashes to its CID.
let mut flipped = car.clone();
let at = flipped.len() - 1;
flipped[at] ^= 1;
let err = open(&flipped, "did:plc:a", &key.public(), |_| true).unwrap_err();
assert!(format!("{err:#}").contains("CID"), "{err:#}");
// Cut short, or no CAR at all.
assert!(open(&car[..car.len() - 3], "did:plc:a", &key.public(), |_| true).is_err());
assert!(open(&[], "did:plc:a", &key.public(), |_| true).is_err());
// A length that never ends.
assert!(open(&[0x80], "did:plc:a", &key.public(), |_| true).is_err());
}
}
crates/knotless-serve/src/firehose.rs+1−0
View file
@@ -59,6 +59,7 @@ pub fn parse(frame: &[u8]) -> Option<Event> {
}
if !event.ops.is_empty() {
event.did = Some(text(&body, "repo")?);
event.rev = text(&body, "rev");
}
}
"#sync" | "#account" | "#identity" => {
crates/knotless-serve/src/index.rs+184−1
View file
@@ -68,6 +68,18 @@ pub struct Index {
stamps: Stamps,
/// Repositories' pulls and issues, as [`Index::assembled`] made them.
kept: Mutex<Kept>,
/// Held across a write weighed by its revision, from the weighing to
/// the stamp, so a backfill and the feed cannot both pass it at once
/// ([`Index::put_as_of`]).
revs: Mutex<()>,
}
/// What backfills made of an account ([`Index::backfilled`]).
#[derive(Debug, Default, PartialEq)]
pub struct Backfilled {
pub rev: Option<String>,
pub tries: u32,
pub retry_at: u64,
}
/// A repository's pulls and issues: the forge records about it that stand,
@@ -520,6 +532,11 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
sealed BLOB NOT NULL, at TEXT NOT NULL,
PRIMARY KEY (owner, repo, runner, name));
CREATE TABLE IF NOT EXISTS inactive (did TEXT PRIMARY KEY, status TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS record_rev (
uri TEXT PRIMARY KEY, did TEXT NOT NULL, rev TEXT NOT NULL, gone INTEGER NOT NULL);
CREATE INDEX IF NOT EXISTS record_rev_gone ON record_rev (did, rev) WHERE gone;
CREATE TABLE IF NOT EXISTS backfill (
did TEXT PRIMARY KEY, rev TEXT, tries INTEGER NOT NULL, retry_at INTEGER NOT NULL);
CREATE INDEX IF NOT EXISTS label_uri ON label (uri);
DROP VIEW IF EXISTS temp.down;
CREATE TABLE IF NOT EXISTS down (subject TEXT PRIMARY KEY);
@@ -621,7 +638,8 @@ const CACHE: &str = "DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge;
DROP TABLE IF EXISTS reaction; DROP TABLE IF EXISTS release; DROP TABLE IF EXISTS mention;
DROP TABLE IF EXISTS reviewer;
DROP TABLE IF EXISTS org; DROP TABLE IF EXISTS org_member; DROP TABLE IF EXISTS membership;
DROP TABLE IF EXISTS collaborator; DROP TABLE IF EXISTS shared;";
DROP TABLE IF EXISTS collaborator; DROP TABLE IF EXISTS shared;
DROP TABLE IF EXISTS record_rev; DROP TABLE IF EXISTS backfill;";
/// A label that has not lapsed.
const LIVE: &str = "(exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))";
@@ -774,6 +792,7 @@ impl Index {
reading: Mutex::new(Some(reading)),
stamps: Stamps::new(STAMPS_MAX),
kept: Mutex::new(Kept::new(KEPT_ENTRIES)),
revs: Mutex::default(),
};
Ok(index)
}
@@ -789,6 +808,122 @@ impl Index {
self.put_at("", did, collection, rkey, value)
}
/// [`Index::put`], of the record as the commit at `rev` left it,
/// unless a later commit's write to it is indexed already: a backfill
/// reads a repository while the feed goes on, and must not put back
/// what the feed has since changed or deleted. With no revision it is
/// [`Index::put`].
pub fn put_as_of(
&self,
did: &str,
collection: &str,
rkey: &str,
value: &Value,
rev: Option<&str>,
) -> rusqlite::Result<()> {
self.as_of(did, collection, rkey, rev, false, &|| {
self.put(did, collection, rkey, value)
})
}
/// [`Index::delete`], as of `rev`, as [`Index::put_as_of`] puts. The
/// deletion is remembered until a backfill reads the account past it
/// ([`Index::read_through`]), so the record is not put back meanwhile.
pub fn delete_as_of(
&self,
did: &str,
collection: &str,
rkey: &str,
rev: Option<&str>,
) -> rusqlite::Result<()> {
self.as_of(did, collection, rkey, rev, true, &|| {
self.delete(did, collection, rkey)
})
}
/// Runs `write` unless the record's last indexed write came at a
/// revision after `rev`, then stamps `rev` on it, `gone` for a deletion.
fn as_of(
&self,
did: &str,
collection: &str,
rkey: &str,
rev: Option<&str>,
gone: bool,
write: &dyn Fn() -> rusqlite::Result<()>,
) -> rusqlite::Result<()> {
let Some(rev) = rev else {
return write();
};
let _held = self.revs.lock().unwrap();
let uri = record_uri("", did, collection, rkey);
let later = self.db.lock().unwrap().query_row(
"SELECT 1 FROM record_rev WHERE uri = ?1 AND rev > ?2",
params![uri, rev],
|_| Ok(()),
);
if later.optional()?.is_some() {
return Ok(());
}
write()?;
self.db.lock().unwrap().execute(
"INSERT INTO record_rev VALUES (?1, ?2, ?3, ?4)
ON CONFLICT (uri) DO UPDATE SET rev = excluded.rev, gone = excluded.gone",
params![uri, did, rev, gone],
)?;
Ok(())
}
/// What backfills made of `did`: the revision one last read it at, and
/// how many have failed since, the last of them asking for no other
/// before `retry_at`, in seconds since the epoch.
pub fn backfilled(&self, did: &str) -> rusqlite::Result<Backfilled> {
let db = self.db.lock().unwrap();
let found = db.query_row(
"SELECT rev, tries, retry_at FROM backfill WHERE did = ?1",
[did],
|row| {
Ok(Backfilled {
rev: row.get(0)?,
tries: row.get(1)?,
retry_at: row.get(2)?,
})
},
);
Ok(found.optional()?.unwrap_or_default())
}
/// Notes that a backfill read `did` as of `rev`.
pub fn backfill_read(&self, did: &str, rev: &str) -> rusqlite::Result<()> {
self.db.lock().unwrap().execute(
"INSERT OR REPLACE INTO backfill VALUES (?1, ?2, 0, 0)",
params![did, rev],
)?;
Ok(())
}
/// Notes that a backfill failed to read `did`, and that none is to try
/// again before `retry_at`.
pub fn backfill_failed(&self, did: &str, retry_at: u64) -> rusqlite::Result<()> {
self.db.lock().unwrap().execute(
"INSERT INTO backfill VALUES (?1, NULL, 1, ?2)
ON CONFLICT (did) DO UPDATE SET tries = tries + 1, retry_at = excluded.retry_at",
params![did, retry_at],
)?;
Ok(())
}
/// Forgets the deletions of `did`'s records made at or before `rev`,
/// once a backfill has read the account as of then: what it read no
/// longer holds them.
pub fn read_through(&self, did: &str, rev: &str) -> rusqlite::Result<()> {
self.db.lock().unwrap().execute(
"DELETE FROM record_rev WHERE did = ?1 AND rev <= ?2 AND gone",
params![did, rev],
)?;
Ok(())
}
/// [`Index::put`], of a record in `space`, or in public records when
/// it is empty: where an organization's record is kept tells what it
/// says apart, as one keeps a record of the same name in each.
@@ -3845,6 +3980,54 @@ mod tests {
json!({"name": "r", "key": "k", "createdAt": "2026-01-01T00:00:00Z"})
}
#[test]
fn a_backfill_never_undoes_a_later_write() {
let dir = tempfile::tempdir().unwrap();
let index = Index::open(&dir.path().join("i.db")).unwrap();
let uri = "at://did:a/dev.knotless.repo/r";
let named = |about: &str| json!({"name": "r", "key": "k", "description": about});
let about = || index.repo(uri).unwrap().map(|r| r["description"].clone());
// The feed's write at rev 3 stands over a backfill's read as of 2.
let (fed, read) = (named("fed"), named("read"));
index
.put_as_of("did:a", nsid::REPO, "r", &fed, Some("3"))
.unwrap();
index
.put_as_of("did:a", nsid::REPO, "r", &read, Some("2"))
.unwrap();
assert_eq!(about(), Some(json!("fed")));
// One as of a later commit replaces it.
index
.put_as_of("did:a", nsid::REPO, "r", &read, Some("4"))
.unwrap();
assert_eq!(about(), Some(json!("read")));
// A deletion at 5 keeps a read as of 4 from putting it back, and
// one from before it changes nothing.
index
.delete_as_of("did:a", nsid::REPO, "r", Some("5"))
.unwrap();
index
.delete_as_of("did:a", nsid::REPO, "r", Some("1"))
.unwrap();
index
.put_as_of("did:a", nsid::REPO, "r", &read, Some("4"))
.unwrap();
assert_eq!(about(), None);
// A read past it forgets it, and what was read before it then lands.
index.read_through("did:a", "4").unwrap();
index
.put_as_of("did:a", nsid::REPO, "r", &read, Some("4"))
.unwrap();
assert_eq!(about(), None);
index.read_through("did:a", "5").unwrap();
index
.put_as_of("did:a", nsid::REPO, "r", &read, Some("4"))
.unwrap();
assert_eq!(about(), Some(json!("read")));
index.delete_as_of("did:a", nsid::REPO, "r", None).unwrap();
assert_eq!(about(), None);
}
#[test]
fn pulls_and_issues_are_kept_until_written_about() {
let dir = tempfile::tempdir().unwrap();
crates/knotless-serve/src/lib.rs+107−26
View file
@@ -17,6 +17,7 @@
pub mod accounts;
mod api;
mod appview;
pub mod archive;
pub mod card;
pub mod diff;
pub mod firehose;
@@ -315,7 +316,7 @@ impl App {
};
let (ctx, index, did) = (self.ctx.clone(), index.clone(), did.to_string());
tokio::spawn(async move {
let _ = index_op(&ctx, &index, &did, &op).await;
let _ = index_op(&ctx, &index, &did, &op, None).await;
});
}
}
@@ -389,15 +390,17 @@ impl App {
.all(|op| op.action == "delete" || op.record.is_some())
{
for op in &event.ops {
telemetry::indexed(&did, index_inline(&index, &did, op));
let rev = event.rev.as_deref();
telemetry::indexed(&did, index_inline(&index, &did, op, rev));
}
accounts::repos_deleted(self, &did, &deleted);
return;
}
let (app, ctx) = (self.clone(), self.ctx.clone());
tokio::spawn(async move {
for op in event.ops {
telemetry::indexed(&did, index_op(&ctx, &index, &did, &op).await);
let rev = event.rev.as_deref();
for op in &event.ops {
telemetry::indexed(&did, index_op(&ctx, &index, &did, op, rev).await);
}
accounts::repos_deleted(&app, &did, &deleted);
});
@@ -612,21 +615,35 @@ impl App {
}
}
/// Indexes a write the event carried in full: a deletion, or a record
/// delivered inline. `false` when the record has to be fetched.
fn index_inline(index: &index::Index, did: &str, op: &firehose::Op) -> rusqlite::Result<bool> {
/// Indexes a write the event carried in full, made in the commit at `rev`
/// when it says: a deletion, or a record delivered inline. `false` when
/// the record has to be fetched.
fn index_inline(
index: &index::Index,
did: &str,
op: &firehose::Op,
rev: Option<&str>,
) -> rusqlite::Result<bool> {
let (collection, rkey) = (op.collection.as_str(), op.rkey.as_str());
match (&op.record, op.action.as_str()) {
(_, "delete") => index.delete(did, &op.collection, &op.rkey)?,
(Some(record), _) => index.put(did, &op.collection, &op.rkey, record)?,
(_, "delete") => index.delete_as_of(did, collection, rkey, rev)?,
(Some(record), _) => index.put_as_of(did, collection, rkey, record, rev)?,
(None, _) => return Ok(false),
}
Ok(true)
}
/// Indexes one write: the record the event carried, else the record as its
/// PDS now has it, or its absence.
pub async fn index_op(ctx: &Ctx, index: &index::Index, did: &str, op: &firehose::Op) -> Result<()> {
if index_inline(index, did, op)? {
/// Indexes one write, made in the commit at `rev` when it is known: the
/// record the event carried, else the record as its PDS now has it, which
/// is as of that commit or later.
pub async fn index_op(
ctx: &Ctx,
index: &index::Index,
did: &str,
op: &firehose::Op,
rev: Option<&str>,
) -> Result<()> {
if index_inline(index, did, op, rev)? {
return Ok(());
}
let id = knotless::identity::resolve(ctx, did).await?;
@@ -641,14 +658,17 @@ pub async fn index_op(ctx: &Ctx, index: &index::Index, did: &str, op: &firehose:
None,
)
.await?;
index.put(did, &op.collection, &op.rkey, &out["value"])?;
index.put_as_of(did, &op.collection, &op.rkey, &out["value"], rev)?;
Ok(())
}
/// Backfills from `service`, saying how it went.
pub async fn backfilled(ctx: &Ctx, index: &index::Index, service: &str) -> String {
match backfill(ctx, index, service).await {
Ok(n) => format!("backfilled {n} records from {service}"),
Ok(f) => format!(
"backfilled {} records of {} accounts from {service}, {} unchanged, {} failed",
f.records, f.read, f.unchanged, f.failed
),
Err(err) => format!("backfilling from {service}: {err:#}"),
}
}
@@ -723,22 +743,83 @@ async fn pages(
Ok(dids)
}
/// What a backfill did: the records it indexed, from the accounts it
/// read; the accounts it found unchanged since it last read them, or put
/// off as one failed not long ago; and those it failed to read.
#[derive(Debug, Default)]
pub struct Filled {
pub records: usize,
pub read: usize,
pub unchanged: usize,
pub failed: usize,
}
/// How many accounts a backfill reads at once.
const READERS: usize = 4;
/// Indexes every account on `service`, a PDS or relay, that holds knotless
/// records: each account's repositories, checks and forge records.
pub async fn backfill(ctx: &Ctx, index: &index::Index, service: &str) -> Result<usize> {
/// records: each account's repositories, checks and forge records, read
/// whole from its PDS and checked ([`archive`]).
///
/// An account whose latest commit is the one last read is passed over. A
/// record the feed indexed from a later commit than the one read stands
/// ([`index::Index::put_as_of`]). An account that fails is tried again
/// by a later backfill, a minute later for each time it failed, up to an
/// hour, and does not stop the rest.
pub async fn backfill(ctx: &Ctx, index: &index::Index, service: &str) -> Result<Filled> {
use futures::StreamExt;
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), service);
let mut indexed = 0;
for did in accounts(&xrpc).await? {
let url = RepoUrl::parse(&format!("knotless://{did}/_"))?;
let remote = Remote::connect(ctx, &url).await?;
for collection in indexed_collections() {
for r in remote.list_all::<serde_json::Value>(collection).await? {
index.put(&did, collection, &r.rkey, &r.value)?;
indexed += 1;
let reads = futures::stream::iter(accounts(&xrpc).await?)
.map(|did| async move {
let read = read_account(ctx, index, &did).await;
(did, read)
})
.buffer_unordered(READERS);
let reads: Vec<(String, Result<Option<usize>>)> = reads.collect().await;
let mut filled = Filled::default();
for (did, read) in reads {
match read {
Ok(Some(records)) => {
(filled.records, filled.read) = (filled.records + records, filled.read + 1)
}
Ok(None) => filled.unchanged += 1,
Err(err) => {
filled.failed += 1;
telemetry::backfill_failed(&did, &err);
let tries = u64::from(index.backfilled(&did)?.tries);
index.backfill_failed(&did, telemetry::now_secs() + 60 * (tries + 1).min(60))?;
}
}
}
Ok(indexed)
Ok(filled)
}
/// Reads `did` into the index, and how many records it held; `None` when
/// it is as last read, or a failure put it off until later.
async fn read_account(ctx: &Ctx, index: &index::Index, did: &str) -> Result<Option<usize>> {
#[derive(serde::Deserialize)]
struct Latest {
rev: String,
}
let was = index.backfilled(did)?;
if was.retry_at > telemetry::now_secs() {
return Ok(None);
}
let id = knotless::identity::resolve(ctx, did).await?;
let xrpc = knotless::xrpc::Xrpc::new(ctx.http.clone(), &id.pds);
let asked = [("did", did)];
let latest = xrpc.query::<Latest>("com.atproto.sync.getLatestCommit", &asked, None);
if was.rev.as_deref() == Some(latest.await?.rev.as_str()) {
return Ok(None);
}
let kept = |c: &str| indexed_collections().any(|i| i == c);
let read = archive::fetch(ctx, did, &id.pds, archive::MAX, kept).await?;
for r in &read.records {
index.put_as_of(did, &r.collection, &r.rkey, &r.value, Some(&read.rev))?;
}
index.read_through(did, &read.rev)?;
index.backfill_read(did, &read.rev)?;
Ok(Some(read.records.len()))
}
/// How often the index is filled again from the relay.
crates/knotless-serve/src/plans.rs+2−1
View file
@@ -32,7 +32,7 @@ pub const ENV: &str = "KNOTLESS_PLANS";
/// The tables that grow with what people do, as many rows as prod might
/// hold: a scan of one is a scan of the network. The rest are said to be
/// small, as a scan of them is.
pub const GROWING: [&str; 16] = [
pub const GROWING: [&str; 17] = [
"forge",
"check_run",
"comment",
@@ -49,6 +49,7 @@ pub const GROWING: [&str; 16] = [
"theme",
"shared",
"shared_use",
"record_rev",
];
/// The rows a growing table is said to hold, and a small one.
crates/knotless-serve/src/telemetry.rs+17−0
View file
@@ -273,6 +273,23 @@ pub fn indexed<T, E: std::fmt::Display>(did: &str, done: Result<T, E>) {
}
}
/// Logs, at warn, why a backfill could not read `did`: a later one tries
/// again.
pub fn backfill_failed(did: &str, err: &anyhow::Error) {
let fields = Fields {
did: Some(did),
error: Some(format!("{err:#}")),
..Fields::default()
};
telemetry::note(Level::WARN, "account not backfilled", &fields);
}
/// Seconds since the epoch, now.
pub fn now_secs() -> u64 {
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH);
now.unwrap_or_default().as_secs()
}
/// Logs, at warn, why a sync of the space `uri` failed, if it did: it is
/// tried again at the next.
pub fn space_synced(uri: &str, done: anyhow::Result<()>) {
crates/knotless-serve/tests/forge.rs+21−5
View file
@@ -343,7 +343,8 @@ fn forks_pulls_reviews_and_merges() {
&av.app.ctx,
index,
&owner.account.did,
&gone
&gone,
None
))
.is_err()
);
@@ -482,7 +483,7 @@ fn forks_pulls_reviews_and_merges() {
.rt
.block_on(knotless_serve::backfill(&ctx(&s), &index, &pds))
.unwrap();
assert!(n >= 10, "{n}");
assert!(n.records >= 10 && n.read == 2, "{n:?}");
let report = av.rt.block_on(knotless_serve::backfilled(
&ctx(&s),
&index,
@@ -492,10 +493,11 @@ fn forks_pulls_reviews_and_merges() {
report.starts_with("backfilling from http://127.0.0.1:9: "),
"{report}"
);
// Read again with nothing new, each account is passed over.
let report = av
.rt
.block_on(knotless_serve::backfilled(&ctx(&s), &index, &pds));
assert!(report.starts_with("backfilled "), "{report}");
assert!(report.ends_with(", 2 unchanged, 0 failed"), "{report}");
let did = owner.account.did.as_str();
let backfilled = index.forge(did, &key).unwrap().len();
wait_until(|| indexed.forge(did, &key).unwrap().len() == backfilled);
@@ -509,7 +511,10 @@ fn forks_pulls_reviews_and_merges() {
let by = axum::routing::get(
move |q: axum::extract::Query<std::collections::HashMap<String, String>>| {
let page = match q.get("cursor") {
None => serde_json::json!({"repos": [{"did": holder}], "cursor": "2"}),
None => serde_json::json!({
"repos": [{"did": holder}, {"did": "did:plc:nobodyhereatall"}],
"cursor": "2"
}),
Some(_) => serde_json::json!({"repos": []}),
};
async move { axum::Json(page) }
@@ -521,9 +526,19 @@ fn forks_pulls_reviews_and_merges() {
});
let fresh = Index::open(&s.path("relay.db")).unwrap();
assert!(fresh.rebuilt);
av.rt
let filled = av
.rt
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay))
.unwrap();
// One it cannot read is tried again later, and does not stop the rest.
assert_eq!((filled.read, filled.failed), (1, 1), "{filled:?}");
let nobody = fresh.backfilled("did:plc:nobodyhereatall").unwrap();
assert_eq!((nobody.rev, nobody.tries), (None, 1));
let again = av
.rt
.block_on(knotless_serve::backfill(&ctx(&s), &fresh, &relay))
.unwrap();
assert_eq!((again.unchanged, again.failed), (2, 0), "{again:?}");
// Only the owner's: what others wrote, the relay here does not name.
assert!(fresh.repo(&uri).unwrap().is_some());
let owners = fresh.forge(did, &key).unwrap().len();
@@ -555,6 +570,7 @@ fn forks_pulls_reviews_and_merges() {
&index,
&contrib.account.did,
&deleted,
None,
))
.unwrap();
assert_eq!(index.forge(did, &key).unwrap().len(), backfilled - 1);
deploy/README.md+1−1
View file
@@ -711,7 +711,7 @@ What serve logs:
- `knotless-serve starting` (`version`, `commit`), `index opened`
(`duration_ms`, `rebuilt`), `listening`, `stopping`, `stopped`;
`feed connected` and `feed disconnected`; `index sweep started` and
`ended`; `event dropped: not indexed`; `panicked`.
`ended`; `event dropped: not indexed`; `account not backfilled`; `panicked`.
What the runner logs: `run queued`, `run started`, `run ended` (`status`,
`duration_ms`), `run cancelled before it started`, each with `repo`,
docs/design/serve.md+19−0
View file