chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A push makes only its own repository's mirrors stale, not the account's
crates/knotless-serve/src/lib.rs+129−5
| @@ -80,6 +80,11 @@ pub struct App { | ||
pub reprove: std::sync::atomic::AtomicU64, | ||
/// Firehose events seen per DID. | ||
events: Mutex<HashMap<String, u64>>, | ||
/// Of those, the ones that could change any of the account's | ||
/// repositories, per DID ([`App::stamp`]). | ||
account_events: Mutex<HashMap<String, u64>>, | ||
/// And those that change one repository, by its owner's DID and key. | ||
repo_events: Mutex<HashMap<(String, String), u64>>, | ||
/// The last commit, by its rev, that wrote to each DID's repository | ||
/// collections ([`first_of`]). | ||
revs: Mutex<HashMap<String, String>>, | ||
| @@ -164,7 +169,8 @@ struct Entry { | ||
/// Where a private repository was then found, read until it is | ||
/// proved again. | ||
remote: Option<Remote>, | ||
events: u64, | ||
/// The events it was last synced as of ([`App::stamp`]). | ||
events: (u64, u64), | ||
/// Its space's moves when it was last synced, for a private one. | ||
moves: u64, | ||
} | ||
| @@ -222,6 +228,8 @@ impl App { | ||
viewed: Mutex::default(), | ||
reprove: std::sync::atomic::AtomicU64::new(private::REPROVE.as_millis() as u64), | ||
events: Mutex::default(), | ||
account_events: Mutex::default(), | ||
repo_events: Mutex::default(), | ||
revs: Mutex::default(), | ||
watching: AtomicBool::new(false), | ||
moves: Mutex::default(), | ||
| @@ -366,9 +374,15 @@ impl App { | ||
// Jetstream sends each record a commit wrote as an event of its own, | ||
// and all of them are there once the first is: a push's pack, claim | ||
// and ref are one sync, not a sync and then another for the rest. | ||
// A record of one repository makes only its mirrors stale; anything | ||
// else, all of the account's. | ||
let moved = event.ops.is_empty() || repository; | ||
if moved && first_of(&self.revs, &did, event.rev.as_deref()) { | ||
self.touch(&did); | ||
let keys = scopes(&event.ops); | ||
for key in keys.iter().filter(|_| moved) { | ||
let seen = format!("{did} {}", key.as_deref().unwrap_or_default()); | ||
if first_of(&self.revs, &seen, event.rev.as_deref()) { | ||
self.heard_of(&did, key.as_deref()); | ||
} | ||
} | ||
if let Some(index) = self.index.clone() | ||
&& !event.ops.is_empty() | ||
| @@ -424,6 +438,40 @@ impl App { | ||
self.events.lock().unwrap().get(did).copied().unwrap_or(0) | ||
} | ||
| ||
/// What a mirror of `did`'s repository with `key` goes stale with: the | ||
/// events that could change any of the account's repositories, and | ||
/// those of that one. One of a key not yet known goes stale with every | ||
/// event of the account. | ||
pub(crate) fn stamp(&self, did: &str, key: Option<&str>) -> (u64, u64) { | ||
let Some(key) = key else { | ||
return (self.events_for(did), u64::MAX); | ||
}; | ||
let account = self.account_events.lock().unwrap().get(did).copied(); | ||
let repos = self.repo_events.lock().unwrap(); | ||
let repo = repos.get(&(did.to_string(), key.to_string())).copied(); | ||
(account.unwrap_or(0), repo.unwrap_or(0)) | ||
} | ||
| ||
/// What `entry`, synced now, is fresh as of: `stamp`, taken before its | ||
/// sync for the key it had then; for a key the sync first found, as | ||
/// of now, but only if no event of the account came meanwhile, as | ||
/// `any`, its events then, says. Otherwise none, to sync again. | ||
pub(crate) fn settled( | ||
&self, | ||
did: &str, | ||
entry_key: Option<&str>, | ||
stamp: (u64, u64), | ||
any: u64, | ||
found: Option<&str>, | ||
) -> (u64, u64) { | ||
if found == entry_key { | ||
return stamp; | ||
} | ||
let quiet = self.events_for(did) == any; | ||
let stale = (u64::MAX, u64::MAX); | ||
[stale, self.stamp(did, found)][usize::from(quiet)] | ||
} | ||
| ||
/// Syncs of the space `uri` that found something written there. | ||
pub fn moves_of(&self, uri: &str) -> u64 { | ||
self.moves.lock().unwrap().get(uri).copied().unwrap_or(0) | ||
| @@ -443,12 +491,33 @@ impl App { | ||
/// Records that `did` wrote to a repository collection, and refreshes its | ||
/// mirrors in the background. | ||
pub fn touch(self: &Arc<App>, did: &str) { | ||
self.heard_of(did, None); | ||
} | ||
| ||
/// Records that `did` wrote to its repository with `key`, or with none | ||
/// to what could change any of them, and refreshes its mirrors in the | ||
/// background: those it did not write to are found fresh, and left. | ||
fn heard_of(self: &Arc<App>, did: &str, key: Option<&str>) { | ||
*self | ||
.events | ||
.lock() | ||
.unwrap() | ||
.entry(did.to_string()) | ||
.or_default() += 1; | ||
match key { | ||
Some(key) => { | ||
let mut repos = self.repo_events.lock().unwrap(); | ||
*repos.entry((did.into(), key.into())).or_default() += 1; | ||
} | ||
None => { | ||
*self | ||
.account_events | ||
.lock() | ||
.unwrap() | ||
.entry(did.into()) | ||
.or_default() += 1 | ||
} | ||
} | ||
let names: Vec<String> = self | ||
.repos | ||
.lock() | ||
| @@ -525,7 +594,9 @@ impl App { | ||
.or_default() | ||
.clone(); | ||
let mut entry = entry.lock().await; | ||
let events = self.events_for(&did); | ||
let key = entry.key(&did); | ||
let any = self.events_for(&did); | ||
let events = self.stamp(&did, key.as_deref()); | ||
let fresh = match (&entry.git, entry.synced) { | ||
(Some(_), Some(_)) if self.watching.load(Ordering::Relaxed) => entry.events == events, | ||
// A write from the pages touches it either way. Not there, it is | ||
| @@ -543,7 +614,8 @@ impl App { | ||
let synced = self.sync(&self.ctx, remote, url, &mut entry).await; | ||
// Found, or found not there, as of these events. | ||
if synced.is_ok() || entry.git.is_none() { | ||
entry.events = events; | ||
let found = entry.key(&did); | ||
entry.events = self.settled(&did, key.as_deref(), events, any, found.as_deref()); | ||
} | ||
synced?; | ||
} | ||
| @@ -642,6 +714,35 @@ pub(crate) fn aside(mirror: &Mirror) { | ||
tokio::spawn(async move { mirror.tidy().await.ok() }); | ||
} | ||
| ||
/// The repositories `ops` write to, by key, each once: a record of one | ||
/// repository, kept under its key, names it; any other, or none at all, | ||
/// is `None`, for all of them. | ||
fn scopes(ops: &[firehose::Op]) -> Vec<Option<String>> { | ||
use knotless::records::nsid; | ||
let one = [nsid::REF, nsid::CLAIM, nsid::PACK]; | ||
let mut keys: Vec<Option<String>> = ops | ||
.iter() | ||
.map(|op| { | ||
let key = op.rkey.split_once('.').map(|(key, _)| key.to_string()); | ||
key.filter(|_| one.contains(&op.collection.as_str())) | ||
}) | ||
.collect(); | ||
if keys.is_empty() { | ||
keys.push(None); | ||
} | ||
keys.sort(); | ||
keys.dedup(); | ||
keys | ||
} | ||
| ||
impl Entry { | ||
/// The key of the repository it last found, if it is still `did`'s. | ||
fn key(&self, did: &str) -> Option<String> { | ||
let found = self.found.as_ref().filter(|(now, _, _)| now == did); | ||
found.map(|(_, key, _)| key.clone()) | ||
} | ||
} | ||
| ||
/// Whether `rev` is a commit of `did`'s that `seen` has not had yet, and | ||
/// from now on has. An event naming no commit always is one. | ||
fn first_of(seen: &Mutex<HashMap<String, String>>, did: &str, rev: Option<&str>) -> bool { | ||
| @@ -1479,6 +1580,29 @@ fn mcp_appview(named: Option<String>, indexed: bool, addr: SocketAddr) -> Option | ||
#[cfg(test)] | ||
mod tests { | ||
| ||
#[test] | ||
fn a_record_of_one_repository_is_its_alone() { | ||
let op = |collection: &str, rkey: &str| firehose::Op { | ||
action: "create".into(), | ||
collection: collection.into(), | ||
rkey: rkey.into(), | ||
record: None, | ||
}; | ||
use knotless::records::nsid; | ||
// A push's pack, claim and ref name one repository, once. | ||
let push = [ | ||
op(nsid::PACK, "k.3kaaa"), | ||
op(nsid::CLAIM, "k.m.1"), | ||
op(nsid::REF, "k.m"), | ||
]; | ||
assert_eq!(scopes(&push), [Some("k".to_string())]); | ||
// A repository's own record, or one under no key, is of them all. | ||
let both = [op(nsid::REF, "j.m"), op(nsid::REPO, "name")]; | ||
assert_eq!(scopes(&both), [None, Some("j".to_string())]); | ||
assert_eq!(scopes(&[op(nsid::REF, "bare")]), [None]); | ||
assert_eq!(scopes(&[]), [None]); | ||
} | ||
| ||
#[test] | ||
fn one_commit_is_one_touch() { | ||
let seen = Mutex::default(); | ||
crates/knotless-serve/src/private.rs+5−2
| @@ -205,7 +205,9 @@ impl App { | ||
}; | ||
let space = entry.space.clone().unwrap_or_default(); | ||
let moves = self.moves_of(&space); | ||
let events = self.events_for(did); | ||
let entry_key = entry.key(did); | ||
let any = self.events_for(did); | ||
let events = self.stamp(did, entry_key.as_deref()); | ||
// A space no one here has a session for is synced only as it is | ||
// read, with what proved its reader a member. | ||
let followed = self.following.load(Ordering::Relaxed) && self.syncer_of(&space).is_some(); | ||
| @@ -224,7 +226,8 @@ impl App { | ||
} | ||
if !fresh && read.is_ok() { | ||
entry.moves = moves; | ||
entry.events = events; | ||
let found = entry.key(did); | ||
entry.events = self.settled(did, entry_key.as_deref(), events, any, found.as_deref()); | ||
} | ||
// Fresh or not, it is synced as whoever read it last: once its | ||
// syncer logs out, the next to read it takes over. A reader with a | ||
docs/design/serve.md+10−5
| @@ -137,11 +137,16 @@ arrive, without counting them toward any freshness. | ||
Either way, an event counts toward freshness only when it can change a | ||
repository: a write to the repository's own collections (`repo`, `ref`, | ||
`claim`, `pack`, `redirect`), or an identity, account or sync event. A new | ||
comment does not mark a mirror stale. A counted event refreshes that account's | ||
known mirrors in the background, so the next read is often already warm. | ||
Jetstream sends each record a commit wrote as an event of its own, and a | ||
push writes its pack, claim and ref in one commit. Only the first event of a | ||
commit counts, by its `rev`. The commit's other records are already on the | ||
comment does not mark a mirror stale. A `ref`, `claim` or `pack` record is | ||
kept under its repository's key, so it marks only that repository's mirrors | ||
stale: a push to one repository leaves the account's others fresh, where | ||
it once synced every one of them at once, each asking the same PDS. The | ||
`repo` and `redirect` records, an identity, account or sync event, and a | ||
write from the pages mark all of the account's mirrors stale. A counted | ||
event refreshes them in the background, so the next read is often already | ||
warm. Jetstream sends each record a commit wrote as an event of its own, | ||
and a push writes its pack, claim and ref in one commit. Only the first | ||
event of a commit counts for each repository it writes, by its `rev`. The commit's other records are already on the | ||
PDS when the first event arrives, so one sync takes in the whole push. | ||
Counting every event made a second sync follow the first, and each read | ||
waited for both. | ||