~ 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 repository's mirror syncs its refs from the index, not by listing its PDS

Closed chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/refs-from-index
crates/knotless-serve/src/index.rs+116−1
View file
@@ -507,6 +507,9 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
let used = db
.prepare("SELECT 1 FROM sqlite_master WHERE name = 'shared_use'")?
.exists([])?;
let refs_kept = db
.prepare("SELECT 1 FROM sqlite_master WHERE name = 'live_ref'")?
.exists([])?;
// A repository is known by its owner and key; its name, and the
// URIs others wrote naming it, can change.
db.execute_batch(&format!(
@@ -657,6 +660,10 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
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 TABLE IF NOT EXISTS live_ref (
uri TEXT PRIMARY KEY, did TEXT NOT NULL, key TEXT NOT NULL, rkey TEXT NOT NULL,
value TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS live_ref_repo ON live_ref (did, key);
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);
@@ -687,6 +694,11 @@ pub(crate) fn connect(path: &Path) -> rusqlite::Result<(Connection, bool)> {
db.execute_batch("ALTER TABLE push ADD COLUMN pusher TEXT")?;
}
db.execute_batch("CREATE INDEX IF NOT EXISTS push_pusher ON push (pusher, created_at)")?;
// Refs, kept from now on: every account is read whole again by the
// next backfill, so that an account's refs are all here once it is.
if !refs_kept {
db.execute("UPDATE backfill SET rev = NULL", [])?;
}
// Comments, by what they are on, from the forge table, once: an
// index from before had them only there. One whose forge table is
// of another shape gets them as they come.
@@ -764,7 +776,8 @@ const CACHE: &str = "DROP TABLE IF EXISTS repo; DROP TABLE IF EXISTS forge;
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 record_rev; DROP TABLE IF EXISTS backfill;";
DROP TABLE IF EXISTS record_rev; DROP TABLE IF EXISTS backfill;
DROP TABLE IF EXISTS live_ref;";
/// A label that has not lapsed.
const LIVE: &str = "(exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))";
@@ -1013,6 +1026,27 @@ impl Index {
Ok(())
}
/// The live refs of `did`'s repository with `key`, each its record's key
/// and its record, once a backfill has read the account whole, so that
/// the index holds every one; none before.
pub fn live_refs(
&self,
did: &str,
key: &str,
) -> rusqlite::Result<Option<Vec<(String, String)>>> {
let db = self.readers.take(&self.db);
let read = db
.prepare_cached("SELECT 1 FROM backfill WHERE did = ?1 AND rev IS NOT NULL")?
.exists([did])?;
if !read {
return Ok(None);
}
let mut stmt =
db.prepare_cached("SELECT rkey, value FROM live_ref WHERE did = ?1 AND key = ?2")?;
let rows = stmt.query_map([did, key], |row| Ok((row.get(0)?, row.get(1)?)))?;
rows.collect::<rusqlite::Result<_>>().map(Some)
}
/// 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.
@@ -1123,6 +1157,9 @@ impl Index {
None if collection == nsid::BLOCK => Some(did.to_string()),
None => None,
};
if collection == nsid::REF {
live_ref(&db, &uri, did, rkey, value, &text)?;
}
// Bound, then checked: a `?` alone on a line counts as a line of its
// own for coverage.
let wrote = if collection == nsid::REPO {
@@ -1454,6 +1491,7 @@ impl Index {
db.execute("DELETE FROM runner_key WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM release WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM shared WHERE uri = ?1", [&uri])?;
db.execute("DELETE FROM live_ref WHERE uri = ?1", [&uri])?;
db.execute(
"DELETE FROM profile_record WHERE did = ?1 AND collection = ?2 AND rkey = ?3",
[did, collection, rkey],
@@ -3912,6 +3950,26 @@ impl Index {
/// The repositories `did`'s record of `collection`, `value`, is about, as
/// `<did>/<key>`: its own, the one it names, or for a ref that moved, its
/// repository and those it is the head of a pull into.
crates/knotless-serve/src/index.rslines 3950 to 3952
review.knotless.dev

These three lines are about's doc comment. The new live_ref was inserted between them and fn about, so about loses its docs and live_ref gets a wrong paragraph on top of its own. Move live_ref above this comment.

crates/knotless-serve/src/index.rslines 3950 to 3952
review.knotless.dev

These three lines are about's doc comment, and they are still above live_ref. Move live_ref and its docs above them so that about keeps its own.

/// Keeps the ref record at `uri` among its repository's live refs while
/// it names a target: a deleted ref keeps its record, with none.
fn live_ref(
db: &Connection,
uri: &str,
did: &str,
rkey: &str,
value: &Value,
text: &str,
) -> rusqlite::Result<usize> {
let key = value["repo"].as_str().unwrap_or_default();
match value["target"].as_str() {
Some(_) => db.execute(
"INSERT OR REPLACE INTO live_ref VALUES (?1, ?2, ?3, ?4, ?5)",
params![uri, did, key, rkey, text],
),
None => db.execute("DELETE FROM live_ref WHERE uri = ?1", [uri]),
}
}
fn about(db: &Connection, did: &str, collection: &str, value: &Value) -> Vec<String> {
let named = |field: &str| serde_json::from_value::<RepoRef>(value[field].clone()).ok();
let one = |r: RepoRef| vec![format!("{}/{}", r.did(), r.key)];
@@ -5733,6 +5791,63 @@ mod tests {
assert_eq!(pushes, 1);
}
#[test]
fn live_refs_once_an_account_is_read_whole() {
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("i.db");
let index = Index::open(&path).unwrap();
let at = |name: &str, target: Option<&str>| {
json!({"repo": "k", "name": name, "target": target, "generation": 1,
"updatedAt": "2026-01-02T00:00:00Z"})
};
index.backfill_read("did:a", "r1").unwrap();
index
.put(
"did:a",
nsid::REF,
"k.m",
&at("refs/heads/main", Some("a1")),
)
.unwrap();
index
.put("did:a", nsid::REF, "k.f", &at("refs/heads/f", Some("b2")))
.unwrap();
index
.put(
"did:a",
nsid::REF,
"j.m",
&at("refs/heads/main", Some("c3")),
)
.unwrap();
let live = || {
let mut found = index.live_refs("did:a", "k").unwrap().unwrap();
found.sort();
found.into_iter().map(|(rkey, _)| rkey).collect::<Vec<_>>()
};
assert_eq!(live(), ["k.f", "k.m"]);
// A deleted ref keeps its record, with no target; a record deleted
// goes with its repository.
index
.put("did:a", nsid::REF, "k.f", &at("refs/heads/f", None))
.unwrap();
assert_eq!(live(), ["k.m"]);
index.delete("did:a", nsid::REF, "k.m").unwrap();
assert!(live().is_empty());
// Of an account no backfill has read whole, the index may not hold
// every ref, so it says none.
index.backfill_failed("did:b", 0).unwrap();
assert_eq!(index.live_refs("did:b", "k").unwrap(), None);
// Refs are kept from an index that kept none before: every account
// is to be read whole again.
drop(index);
let db = Connection::open(&path).unwrap();
db.execute_batch("DROP TABLE live_ref").unwrap();
drop(db);
let index = Index::open(&path).unwrap();
assert_eq!(index.live_refs("did:a", "k").unwrap(), None);
}
#[test]
fn pushers() {
let dir = tempfile::tempdir().unwrap();
crates/knotless-serve/src/lib.rs+92−6
View file
@@ -80,6 +80,10 @@ pub struct App {
pub reprove: std::sync::atomic::AtomicU64,
/// Firehose events seen per DID.
events: Mutex<HashMap<String, u64>>,
/// When the pages last wrote for each account, which a feed may not
/// have told the index of yet: until [`UNHEARD_FOR`] after, a sync of
/// a mirror of theirs reads their PDS.
unheard: Mutex<HashMap<String, Instant>>,
pub watching: AtomicBool,
/// Syncs of each space that found something written there, by URI.
moves: Mutex<HashMap<String, u64>>,
@@ -224,6 +228,7 @@ impl App {
viewed: Mutex::default(),
reprove: std::sync::atomic::AtomicU64::new(private::REPROVE.as_millis() as u64),
events: Mutex::default(),
unheard: Mutex::default(),
watching: AtomicBool::new(false),
moves: Mutex::default(),
following: AtomicBool::new(false),
@@ -364,9 +369,18 @@ impl App {
.ops
.iter()
.any(|op| knotless::records::nsid::ALL.contains(&op.collection.as_str()));
if event.ops.is_empty() || repository {
self.touch(&did);
// A repository's record and its refs are what its mirrors sync to,
// from the index: read from the PDS, not believed from a frame no
// one signed, as Jetstream's are.
for op in &mut event.ops {
let (repo, r#ref) = (knotless::records::nsid::REPO, knotless::records::nsid::REF);
if [repo, r#ref].contains(&op.collection.as_str()) {
op.record = None;
}
}
// Its mirrors are refreshed once the index has what it wrote, as a
// sync reads their refs from there.
let moved = event.ops.is_empty() || repository;
if let Some(index) = self.index.clone()
&& !event.ops.is_empty()
{
@@ -395,6 +409,7 @@ impl App {
telemetry::indexed(&did, index_inline(&index, &did, op, rev));
}
accounts::repos_deleted(self, &did, &deleted);
self.refresh_if(moved, &did);
return;
}
let (app, ctx) = (self.clone(), self.ctx.clone());
@@ -404,7 +419,17 @@ impl App {
telemetry::indexed(&did, index_op(&ctx, &index, &did, op, rev).await);
}
accounts::repos_deleted(&app, &did, &deleted);
app.refresh_if(moved, &did);
});
return;
}
self.refresh_if(moved, &did);
}
/// [`App::refresh_mirrors`], if `moved`.
fn refresh_if(self: &Arc<App>, moved: bool, did: &str) {
if moved {
self.refresh_mirrors(did);
}
}
@@ -437,9 +462,20 @@ impl App {
.or_default() += 1;
}
/// Records that `did` wrote to a repository collection, and refreshes its
/// mirrors in the background.
/// Records that the pages wrote to a repository collection of `did`'s,
/// and refreshes its mirrors in the background, from its PDS, as a
/// feed may not have brought the write to the index yet.
pub fn touch(self: &Arc<App>, did: &str) {
let mut unheard = self.unheard.lock().unwrap();
unheard.retain(|_, at| at.elapsed() < UNHEARD_FOR);
unheard.insert(did.to_string(), Instant::now());
drop(unheard);
self.refresh_mirrors(did);
}
/// Records that `did` wrote to a repository collection, as a feed
/// told the index, and refreshes its mirrors in the background.
fn refresh_mirrors(self: &Arc<App>, did: &str) {
*self
.events
.lock()
@@ -570,7 +606,12 @@ impl App {
entry: &mut Entry,
) -> Result<(), AppError> {
let started = Instant::now();
let snap = remote.snapshot(ctx).await.map_err(AppError::upstream)?;
let kept = self.kept_snapshot(&remote, url);
let from_index = kept.is_some();
let snap = match kept {
Some(snap) => snap,
None => remote.snapshot(ctx).await.map_err(AppError::upstream)?,
};
let (remote, snap) = remote.follow(ctx, snap).await.map_err(AppError::upstream)?;
let records = started.elapsed();
let Some(repo) = &snap.repo else {
@@ -594,7 +635,8 @@ impl App {
entry.found = Some((remote.did().to_string(), key.to_string(), repo.rkey.clone()));
entry.record = repo.value.clone();
entry.space = remote.space.as_ref().map(|s| s.uri.clone());
entry.storage = knotless::storage::of_snapshot(&snap);
// Read from the index, it lists no packs, so says nothing of them.
entry.storage = knotless::storage::of_snapshot(&snap).filter(|_| !from_index);
entry.git = Some(mirror.git);
entry.synced = Some(Instant::now());
// Every read of the repository waits on its sync: how long each
@@ -611,6 +653,45 @@ impl App {
Ok(())
}
/// `remote`'s repository as the index has it, rather than as its PDS
/// lists it, which for a repository of many refs takes a request a page
/// of a hundred, one after another: its record and its live refs. Only
/// a public repository's, while a feed keeps the index current, of an
/// account a backfill read whole, so that the index holds every ref,
/// and not the next after the pages wrote for it ([`App::touch`]). Its
/// packs are left for [`knotless::fetch`] to list, as it does only when
/// the refs name commits the mirror lacks.
fn kept_snapshot(&self, remote: &Remote, url: &RepoUrl) -> Option<knotless::remote::Snapshot> {
use knotless::remote::{Snapshot, Stored};
let did = remote.did();
let unheard = self.unheard.lock().unwrap().get(did).copied();
let unheard = unheard.is_some_and(|at| at.elapsed() < UNHEARD_FOR);
let public = remote.space.is_none() && !unheard;
let watching = self.watching.load(Ordering::Relaxed) && public;
let index = self.index.as_ref().filter(|_| watching)?;
let uri = knotless::forge::repo_uri(did, &url.name);
let value = index.repo(&uri).ok().flatten()?;
let record: RepoRecord = serde_json::from_value(value).ok()?;
let refs = index.live_refs(did, &record.key).ok().flatten()?;
let stored = |(rkey, value): (String, String)| {
let value = serde_json::from_str(&value).ok()?;
let cid = String::new();
Some(Stored { rkey, cid, value })
};
let refs = refs.into_iter().filter_map(stored).collect();
let repo = Stored {
rkey: url.name.clone(),
cid: String::new(),
value: record,
};
Some(Snapshot {
repo: Some(repo),
refs,
packs: knotless::remote::Page::default(),
read_at: None,
})
}
/// What a synced `entry` holds, asked for under `did`, read with
/// `reader`.
fn found(&self, entry: &Entry, did: &str, reader: Ctx) -> Located {
@@ -630,6 +711,11 @@ impl App {
}
}
/// How long after the pages write for an account its mirrors are synced
/// from its PDS rather than the index: longer than a feed takes to bring
/// the write.
const UNHEARD_FOR: Duration = Duration::from_secs(30);
/// Tidies `mirror` in a task of its own, out of the way of the request
/// that synced it and of every other waiting on that sync.
pub(crate) fn aside(mirror: &Mirror) {
crates/knotless-serve/tests/web.rs+31−0
View file
@@ -3212,6 +3212,37 @@ fn the_releases() {
assert_eq!(gone.status().as_u16(), 404);
}
/// A repository's mirror syncs from the refs the index keeps, once a
/// backfill has read its account whole, rather than listing them from its
/// PDS; and right after the pages write for the account, from its PDS.
#[test]
fn a_mirror_syncs_from_the_refs_the_index_keeps() {
let site = serve(true);
let owner = person(account(), &site);
let url = owner.url("kept");
ok(&owner, &["create", &url]);
let work = owner.work("w", 1);
owner.git(&work, &["push", "-q", &url, "main"]);
let did = owner.account.did.clone();
let uri = format!("at://{did}/dev.knotless.repo/kept");
let names = || {
let listed = site.json(&xrpc("listBranches", &[("repo", &uri)]));
let all = listed["branches"].as_array().unwrap().iter();
all.map(|b| b["name"].as_str().unwrap().to_string())
.collect::<Vec<_>>()
};
wait_until(|| names() == ["main"]);
let index = site.app.index.clone().unwrap();
index.backfill_read(&did, "r").unwrap();
owner.git(&work, &["push", "-q", &url, "main:side"]);
wait_until(|| names() == ["main", "side"]);
// Deleted, a branch keeps its record, with no target, and goes.
owner.git(&work, &["push", "-q", &url, ":side"]);
wait_until(|| names() == ["main"]);
site.app.touch(&did);
assert_eq!(names(), ["main"]);
}
#[test]
fn badges_count_where_an_account_lives() {
use knotless_serve::index::Home;
docs/design/serve.md+20−4
View file
@@ -105,6 +105,19 @@ question is when to sync:
When the last feed disconnects, every mirror drops back to the TTL, so the
server never relies on a feed that is not there.
A sync reads the repository's record, its refs and the newest page of its
packs. Its PDS lists refs a hundred to a request, one request after
another, and a deleted ref keeps its record ([storage.md](storage.md)), so a
repository that has merged many pulls takes a request for each hundred
branches it ever had, a second or more, and every read waits on it. So
while a feed is connected, a public repository's record and live refs are
read from the index instead, once a backfill has read its owner's account
whole ([Filling and rebuilding](#filling-and-rebuilding)), which is when
the index holds every one of its refs. Its packs are listed from the PDS
only when the refs name commits the mirror lacks, as after a push. For
thirty seconds after the pages write for an account, before a feed may
have brought the write to the index, its mirrors sync from its PDS.
## Feeds
A feed is a public stream of record changes across the network. The server
@@ -116,9 +129,10 @@ Jetstream (`--jetstream`) is a relay that sends JSON. The server asks it for
traffic, and each event
carries its record decoded inline. Its frames are unsigned: a reader cannot
prove an event is real. The server gives Jetstream the same trust it gives any
relay it chose, and that trust is bounded. The index is a cache, and mirrors
still sync from the PDS, so a false event can at worst cause a needless sync or
a wrong index row.
relay it chose, and that trust is bounded. The index is a cache, and a
repository's record and refs, which its mirrors sync to, are read from its
PDS whichever feed tells of them, never taken from a frame, so a false event
can at worst cause a needless sync or a wrong index row of another kind.
The firehose (`--firehose`) is `com.atproto.sync.subscribeRepos`, from a relay
or a single PDS. It sends every commit on the network as CBOR, and an operation
@@ -474,7 +488,9 @@ column added to the push table is added in place. They
go only when their repository or their owner's account is deleted ([A
repository deleted](#a-repository-deleted)). A
rebuilt index otherwise starts empty, and a feed only brings what happens
next, so it needs a backfill.
next, so it needs a backfill. An index from before refs were kept reads
every account whole again, as a mirror syncs from the index only once it
has.
A backfill reads the knotless records of every account that holds any. A relay
can name those accounts by collection; a PDS lists all of its own. A collection