~ 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 never puts back what the feed has since changed or deleted

Closed chadtmiller.com asked to merge into main from chadtmiller.com/knotless:serve/backfill-revs
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/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+135−1
View file
@@ -68,6 +68,10 @@ 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<()>,
}
/// A repository's pulls and issues: the forge records about it that stand,
@@ -520,6 +524,9 @@ 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 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 +628,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;";
/// A label that has not lapsed.
const LIVE: &str = "(exp IS NULL OR exp > strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))";
@@ -774,6 +782,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 +798,83 @@ 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(())
}
/// 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 +3931,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+45−15
View file
@@ -315,7 +315,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 +389,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 +614,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,7 +657,7 @@ 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(())
}
@@ -725,18 +741,32 @@ async fn pages(
/// Indexes every account on `service`, a PDS or relay, that holds knotless
/// records: each account's repositories, checks and forge records.
///
/// Each account is read as of its latest commit, asked before its records
/// are listed: what the listing finds is that commit's or later, and a
/// write the feed indexed from a later commit stands over it.
pub async fn backfill(ctx: &Ctx, index: &index::Index, service: &str) -> Result<usize> {
#[derive(serde::Deserialize)]
struct Latest {
rev: String,
}
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?;
let asked = [("did", did.as_str())];
let latest = remote
.xrpc
.query::<Latest>("com.atproto.sync.getLatestCommit", &asked, None);
let rev = latest.await?.rev;
for collection in indexed_collections() {
for r in remote.list_all::<serde_json::Value>(collection).await? {
index.put(&did, collection, &r.rkey, &r.value)?;
index.put_as_of(&did, collection, &r.rkey, &r.value, Some(&rev))?;
indexed += 1;
}
}
index.read_through(&did, &rev)?;
}
Ok(indexed)
}
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/tests/forge.rs+3−1
View file
@@ -343,7 +343,8 @@ fn forks_pulls_reviews_and_merges() {
&av.app.ctx,
index,
&owner.account.did,
&gone
&gone,
None
))
.is_err()
);
@@ -555,6 +556,7 @@ fn forks_pulls_reviews_and_merges() {
&index,
&contrib.account.did,
&deleted,
None,
))
.unwrap();
assert_eq!(index.forge(did, &key).unwrap().len(), backfilled - 1);
docs/design/serve.md+9−0
View file
@@ -453,6 +453,15 @@ runs one at startup. `--relay` runs one when the index was rebuilt, then sweeps
every 10 minutes, since a record whose event a feed never delivered would
otherwise be missing for good.
A backfill reads while the feed goes on, so it must not put back what the feed
has since changed. Each write the index takes from a feed keeps the revision of
the commit that made it, and a deletion is remembered with its revision too. A
backfill asks each account for its latest commit before listing its records,
and what it lists is as of that revision or later. A record whose indexed write
came at a later revision is left as the feed has it. Once the account is read,
the deletions at or before that revision are forgotten, as what was read no
longer holds them.
A rollout runs a new build beside the old one, both holding the index open.
That is safe only if the new build would not rebuild it, so the deploy script
asks first with `--check-index`.