chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
Patching one written record into a repository's kept pulls and issues
crates/knotless-serve/src/index.rs+262−17
| @@ -238,8 +238,15 @@ struct Stamps { | ||
/// When everything was last stamped. | ||
all: AtomicU64, | ||
by: Mutex<HashMap<String, u64>>, | ||
/// The records of a repository written since, by scope, each with | ||
/// when: what [`Index::assembled`] reads again to patch what it kept. | ||
items: Mutex<HashMap<String, Vec<(u64, String)>>>, | ||
} | ||
| ||
/// The most records noted as changed in one repository before the whole of | ||
/// it is stamped changed instead: a read patches that many, or builds anew. | ||
const ITEMS_MAX: usize = 32; | ||
| ||
impl Stamps { | ||
fn new(max: usize) -> Stamps { | ||
Stamps { | ||
| @@ -247,6 +254,7 @@ impl Stamps { | ||
clock: AtomicU64::default(), | ||
all: AtomicU64::default(), | ||
by: Mutex::default(), | ||
items: Mutex::default(), | ||
} | ||
} | ||
| ||
| @@ -266,6 +274,38 @@ impl Stamps { | ||
by.insert(scope.to_string(), at); | ||
} | ||
| ||
/// Stamps `uri`, one record of `scope`, as changed, once its write is | ||
/// done: the scope's own stamp stays, so what was kept of it is patched | ||
/// with that record, not built anew. Past [`ITEMS_MAX`] of them, or | ||
/// `max` scopes, it is stamped as a whole. | ||
fn mark_item(&self, scope: &str, uri: &str) { | ||
let at = self.clock.fetch_add(1, Ordering::SeqCst) + 1; | ||
let mut items = self.items.lock().unwrap(); | ||
if items.len() >= self.max && !items.contains_key(scope) { | ||
items.clear(); | ||
self.all.fetch_max(at, Ordering::SeqCst); | ||
} | ||
let noted = items.entry(scope.to_string()).or_default(); | ||
noted.push((at, uri.to_string())); | ||
if noted.len() > ITEMS_MAX { | ||
items.remove(scope); | ||
drop(items); | ||
self.mark(scope); | ||
} | ||
} | ||
| ||
/// The records of `scopes` changed after `after`, each once. | ||
fn touched(&self, scopes: &[String], after: u64) -> Vec<String> { | ||
let items = self.items.lock().unwrap(); | ||
let mut found: Vec<String> = Vec::new(); | ||
for (at, uri) in scopes.iter().filter_map(|s| items.get(s)).flatten() { | ||
if *at > after && !found.contains(uri) { | ||
found.push(uri.clone()); | ||
} | ||
} | ||
found | ||
} | ||
| ||
/// Stamps everything as changed. | ||
fn mark_all(&self) { | ||
let at = self.clock.fetch_add(1, Ordering::SeqCst) + 1; | ||
| @@ -299,6 +339,9 @@ struct Made { | ||
/// The scopes it was read from: the repository, the one it moved | ||
/// from, and its owner's blocks. | ||
scopes: Vec<String>, | ||
/// The owner's hides, which moderating takes out of the records, for | ||
/// a patch to moderate again with. | ||
hides: Vec<Entry>, | ||
/// The record it was assembled with. | ||
record: RepoRecord, | ||
assembled: Arc<Assembled>, | ||
| @@ -1429,7 +1472,13 @@ impl Index { | ||
self.stamps.mark_all(); | ||
} | ||
if let Some(whose) = whose { | ||
self.stamps.mark(&whose); | ||
// A record that is only one of a repository's items, with its | ||
// own comments, changes that one for the next read to patch. | ||
match collection == nsid::REPO || collection == nsid::BLOCK || collection == nsid::HIDE | ||
{ | ||
true => self.stamps.mark(&whose), | ||
false => self.stamps.mark_item(&whose, &uri), | ||
} | ||
} | ||
// A ref that moved, or the record naming the default branch, has | ||
// what the repository is written in counted again, if it was before. | ||
| @@ -1581,9 +1630,17 @@ impl Index { | ||
self.stamps.mark_all(); | ||
} | ||
let blocks = (blocked > 0).then(|| did.to_string()); | ||
for whose in [repo, said, blocks].into_iter().flatten() { | ||
for whose in [repo, blocks].into_iter().flatten() { | ||
self.stamps.mark(&whose); | ||
} | ||
// A hide taken back shows what it hid; any other record leaves | ||
// only itself to read again. | ||
if let Some(whose) = said { | ||
match collection == nsid::HIDE { | ||
true => self.stamps.mark(&whose), | ||
false => self.stamps.mark_item(&whose, &uri), | ||
} | ||
} | ||
Ok(()) | ||
} | ||
| ||
| @@ -2821,6 +2878,13 @@ impl Index { | ||
/// taken down, not by an account taken down, and not kept out by the | ||
/// repository's owner. | ||
pub fn forge(&self, did: &str, key: &str) -> rusqlite::Result<Vec<Entry>> { | ||
let (entries, blocked) = self.forge_read(did, key)?; | ||
Ok(knotless::forge::moderate(did, key, &blocked, entries)) | ||
} | ||
| ||
/// [`Index::forge`] before the owner's hides and blocks are applied, | ||
/// with the accounts they have blocked. | ||
fn forge_read(&self, did: &str, key: &str) -> rusqlite::Result<(Vec<Entry>, Vec<String>)> { | ||
let db = self.readers.take(&self.db); | ||
// With those about the repository it moved from, when both owners | ||
// say so: it names that one, and that one's owner said it moved here. | ||
| @@ -2844,7 +2908,7 @@ impl Index { | ||
let blocked = stmt | ||
.query_map([did], |row| row.get(0))? | ||
.collect::<rusqlite::Result<Vec<String>>>()?; | ||
Ok(knotless::forge::moderate(did, key, &blocked, entries)) | ||
Ok((entries, blocked)) | ||
} | ||
| ||
/// The pulls and issues of `did`'s repository `key`, as `record` says | ||
| @@ -2859,33 +2923,68 @@ impl Index { | ||
record: &RepoRecord, | ||
) -> rusqlite::Result<Arc<Assembled>> { | ||
let name = format!("{did}/{key}"); | ||
// The clock first, before what is kept is judged and anything is | ||
// read, so a write that lands meanwhile stamps past what is made. | ||
let at = self.stamps.now(); | ||
let fresh = |m: &&Made| { | ||
m.record == *record | ||
&& m.made.elapsed() < KEPT_FOR | ||
&& self.stamps.since(&m.scopes) <= m.at | ||
}; | ||
let kept = self.kept.lock().unwrap(); | ||
let found = kept | ||
.by | ||
.get(&name) | ||
.filter(fresh) | ||
.map(|m| m.assembled.clone()); | ||
let found = kept.by.get(&name).filter(fresh).map(|m| { | ||
( | ||
m.assembled.clone(), | ||
m.at, | ||
m.made, | ||
m.scopes.clone(), | ||
m.hides.clone(), | ||
) | ||
}); | ||
drop(kept); | ||
if let Some(found) = knotless::telemetry::looked("assembled", found) { | ||
// Only records of it written since: those are read again, and the | ||
// rest stands, where a write to the repository would have it all | ||
// read and assembled anew. | ||
let touched = found | ||
.as_ref() | ||
.map(|(_, at, _, scopes, _)| self.stamps.touched(scopes, *at)) | ||
.unwrap_or_default(); | ||
let standing = found | ||
.as_ref() | ||
.filter(|_| touched.is_empty()) | ||
.map(|f| f.0.clone()); | ||
if let Some(found) = knotless::telemetry::looked("assembled", standing) { | ||
return Ok(found); | ||
} | ||
// Read after the clock, so a write that lands meanwhile stamps | ||
// past what is kept. | ||
let at = self.stamps.now(); | ||
let from = self.moved_from(did, key)?; | ||
let entries = self.forge(did, key)?; | ||
let (entries, made, scopes, hides) = match found { | ||
Some((was, _, made, scopes, hides)) => { | ||
let entries = self.patched(did, key, &was.entries, &hides, &touched, &scopes)?; | ||
(entries, made, scopes, hides) | ||
} | ||
None => { | ||
let from = self.moved_from(did, key)?; | ||
let (entries, blocked) = self.forge_read(did, key)?; | ||
// Kept apart: moderating takes them out of the entries. | ||
let hide = format!("/{}/", nsid::HIDE); | ||
let is_hide = |e: &&Entry| e.uri.contains(&hide); | ||
let hides: Vec<Entry> = entries.iter().filter(is_hide).cloned().collect(); | ||
let entries = knotless::forge::moderate(did, key, &blocked, entries); | ||
let scopes = [Some(name.clone()), from, Some(did.to_string())]; | ||
( | ||
entries, | ||
Instant::now(), | ||
scopes.into_iter().flatten().collect(), | ||
hides, | ||
) | ||
} | ||
}; | ||
let items = knotless::forge::assemble(record, did, &entries); | ||
let assembled = Arc::new(Assembled { entries, items }); | ||
let scopes = [Some(name.clone()), from, Some(did.to_string())]; | ||
let made = Made { | ||
at, | ||
made: Instant::now(), | ||
scopes: scopes.into_iter().flatten().collect(), | ||
made, | ||
scopes, | ||
hides, | ||
record: record.clone(), | ||
assembled: assembled.clone(), | ||
}; | ||
| @@ -2893,6 +2992,48 @@ impl Index { | ||
Ok(assembled) | ||
} | ||
| ||
/// `was`, the records of `did`'s repository `key` as they stood, with | ||
/// those of `uris` read again: replaced, added, or gone. `scopes` are | ||
/// what `was` was read from, whose last is the owner's blocks, and a | ||
/// record read that is about none of the others is not one of it. | ||
/// Hides and blocks are never among `uris` (a write of one stamps the | ||
/// repository whole), and `hides` are moderated with the rest, so what | ||
/// was moderated before stays so. | ||
fn patched( | ||
&self, | ||
did: &str, | ||
key: &str, | ||
was: &[Entry], | ||
hides: &[Entry], | ||
uris: &[String], | ||
scopes: &[String], | ||
) -> rusqlite::Result<Vec<Entry>> { | ||
let db = self.readers.take(&self.db); | ||
let mut stmt = db.prepare_cached( | ||
"SELECT f.uri, f.author, f.value, f.repo_did || '/' || f.repo_key | ||
FROM json_each(?1) j JOIN forge f ON f.uri = j.value | ||
WHERE f.uri NOT IN down AND f.author NOT IN down | ||
AND coalesce(json_extract(f.value, '$.subject'), '') NOT IN down", | ||
)?; | ||
let read = |row: &rusqlite::Row| Ok((entry(row)?, row.get::<_, String>(3)?)); | ||
let again = stmt | ||
.query_map([serde_json::to_string(uris).unwrap_or_default()], read)? | ||
.collect::<rusqlite::Result<Vec<_>>>()?; | ||
let mut stmt = db.prepare_cached("SELECT subject FROM block WHERE did = ?1")?; | ||
let blocked = stmt | ||
.query_map([did], |row| row.get(0))? | ||
.collect::<rusqlite::Result<Vec<String>>>()?; | ||
let ours = |scope: &String| scopes[..scopes.len() - 1].contains(scope); | ||
let mut entries: Vec<Entry> = was | ||
.iter() | ||
.filter(|e| !uris.contains(&e.uri)) | ||
.cloned() | ||
.collect(); | ||
entries.extend(again.into_iter().filter(|(_, s)| ours(s)).map(|(e, _)| e)); | ||
entries.extend(hides.iter().cloned()); | ||
Ok(knotless::forge::moderate(did, key, &blocked, entries)) | ||
} | ||
| ||
/// The repository `did`'s `key` moved from, as `<did>/<key>`, whose | ||
/// pulls and issues [`Index::forge`] reads with its own. | ||
fn moved_from(&self, did: &str, key: &str) -> rusqlite::Result<Option<String>> { | ||
| @@ -4937,6 +5078,109 @@ mod tests { | ||
assert!(!kept(&mut was, &record)); | ||
} | ||
| ||
#[test] | ||
fn a_record_written_patches_what_is_kept_of_its_repository() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
let index = Index::open(&dir.path().join("i.db")).unwrap(); | ||
index.put("did:a", nsid::REPO, "r", &owned()).unwrap(); | ||
let record: RepoRecord = serde_json::from_value(owned()).unwrap(); | ||
let repo = json!({"uri": "at://did:a/dev.knotless.repo/r", "key": "k"}); | ||
let issue = |title: &str| json!({"repo": repo, "title": title, "createdAt": "2026-01-02T00:00:00Z"}); | ||
index.put("did:b", nsid::ISSUE, "1", &issue("One")).unwrap(); | ||
index.put("did:b", nsid::ISSUE, "2", &issue("Two")).unwrap(); | ||
// The owner blocking another account, whose writes are left out. | ||
let block = json!({"subject": "did:e", "createdAt": "2026-01-03T00:00:00Z"}); | ||
index.put("did:a", nsid::BLOCK, "e", &block).unwrap(); | ||
let was = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(was.items.len(), 2); | ||
// What the whole read gives, to hold a patch to. | ||
let same = |now: &Assembled| { | ||
let sorted = |mut e: Vec<Entry>| { | ||
e.sort_by(|a, b| a.uri.cmp(&b.uri)); | ||
e | ||
}; | ||
let all = index.forge("did:a", "k").unwrap(); | ||
assert_eq!(sorted(now.entries.clone()), sorted(all)); | ||
}; | ||
// One edited, one added, and a comment, patched in. | ||
index.put("did:b", nsid::ISSUE, "1", &issue("Uno")).unwrap(); | ||
index | ||
.put("did:b", nsid::ISSUE, "3", &issue("Three")) | ||
.unwrap(); | ||
let said = json!({"repo": repo, "subject": "at://did:b/dev.knotless.issue/1", | ||
"body": "hi", "createdAt": "2026-01-03T00:00:00Z"}); | ||
index.put("did:b", nsid::COMMENT, "c", &said).unwrap(); | ||
index | ||
.put("did:e", nsid::ISSUE, "1", &issue("Blocked")) | ||
.unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert!(!Arc::ptr_eq(&was, &now)); | ||
assert_eq!(now.items.len(), 3); | ||
assert_eq!(now.entries.len(), 4); | ||
same(&now); | ||
// One deleted, and the same twice. | ||
index.delete("did:b", nsid::ISSUE, "2").unwrap(); | ||
index.delete("did:b", nsid::ISSUE, "2").unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 2); | ||
same(&now); | ||
// One of another repository's, written between, is no part of it. | ||
let other = json!({"uri": "at://did:c/dev.knotless.repo/x", "key": "x"}); | ||
let there = json!({"repo": other, "title": "T", "createdAt": "2026-01-03T00:00:00Z"}); | ||
index.put("did:c", nsid::ISSUE, "9", &there).unwrap(); | ||
assert!(Arc::ptr_eq( | ||
&now, | ||
&index.assembled("did:a", "k", &record).unwrap() | ||
)); | ||
// A hide hides the issue, and taking it back shows it. | ||
let hide = json!({"repo": repo, "subject": "at://did:b/dev.knotless.issue/3", | ||
"createdAt": "2026-01-04T00:00:00Z"}); | ||
index.put("did:a", nsid::HIDE, "h", &hide).unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 1); | ||
index | ||
.put("did:b", nsid::ISSUE, "3", &issue("Three!")) | ||
.unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 1, "still hidden when written again"); | ||
index.delete("did:a", nsid::HIDE, "h").unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 2); | ||
same(&now); | ||
// A write by an author gone inactive is left out, and a flood of | ||
// them is read whole. | ||
index.set_status("did:d", "deactivated").unwrap(); | ||
index | ||
.put("did:d", nsid::ISSUE, "1", &issue("Gone")) | ||
.unwrap(); | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 2); | ||
for n in 0..=ITEMS_MAX { | ||
let rkey = format!("m{n}"); | ||
index.put("did:b", nsid::ISSUE, &rkey, &issue("M")).unwrap(); | ||
} | ||
let now = index.assembled("did:a", "k", &record).unwrap(); | ||
assert_eq!(now.items.len(), 2 + ITEMS_MAX + 1); | ||
same(&now); | ||
} | ||
| ||
#[test] | ||
fn stamps_of_records_note_each_once_and_forget_past_their_number() { | ||
let stamps = Stamps::new(2); | ||
let scopes = ["a".to_string()]; | ||
stamps.mark_item("a", "u"); | ||
stamps.mark_item("a", "u"); | ||
stamps.mark_item("a", "v"); | ||
assert_eq!(stamps.touched(&scopes, 0), ["u", "v"]); | ||
assert!(stamps.touched(&scopes, stamps.now()).is_empty()); | ||
assert_eq!(stamps.since(&scopes), 0); | ||
// Two more repositories, past the number of them. | ||
stamps.mark_item("b", "w"); | ||
stamps.mark_item("c", "x"); | ||
assert!(stamps.since(&scopes) > 0); | ||
assert!(stamps.touched(&scopes, 0).is_empty()); | ||
} | ||
| ||
#[test] | ||
fn pulls_and_issues_of_a_repository_moved_here_are_kept_with_it() { | ||
let dir = tempfile::tempdir().unwrap(); | ||
| @@ -4960,6 +5204,7 @@ mod tests { | ||
at: 0, | ||
made: Instant::now(), | ||
scopes: Vec::new(), | ||
hides: Vec::new(), | ||
record: RepoRecord::default(), | ||
assembled: Arc::new(Assembled { | ||
entries: vec![ | ||
docs/design/serve.md+10−0
| @@ -306,6 +306,16 @@ Details: | ||
private repository, which its owner's settings list, is read from its | ||
record, whose `readers` and `runners` sharing it writes, rather than | ||
asked of its space: the client shows the appview each such write. | ||
- A repository's pulls and issues are assembled from every record said about | ||
it, which on a busy repository takes a good part of a tenth of a second, so | ||
the assembled set is kept in memory by repository for a minute. A write of | ||
one pull, issue or comment (or its deletion) notes only that record against | ||
the repository, and the next read reads just those rows again, replaces | ||
them in what was kept, applies the owner's hides and blocks and assembles | ||
again, instead of reading the whole repository. More than 32 records | ||
written since, a write of the repository's own record, a hide or a block | ||
(or taking one back), a takedown or a label, a move, or the minute passing | ||
(for a label to run out) read it whole. | ||
- A repository has three Atom feeds, for feed readers: `commits.atom`, the | ||
newest 30 commits on its default branch; `tags.atom`, its newest 30 tags | ||
that name a commit; and `releases.atom`, its newest 30 releases, with | ||