chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A push is one sync of its mirrors, not one for each record it wrote
crates/knotless/src/mirror.rs+9−6
| @@ -185,8 +185,8 @@ mod tests { | ||
mirror.tidy().await.unwrap(); | ||
// Another holds the tidy's lock: this one does nothing, and waits | ||
// for nothing. | ||
let held = Lock::try_acquire(mirror.git.dir.with_extension("tidy")).unwrap(); | ||
assert!(held.is_some()); | ||
let held = Lock::acquire(mirror.git.dir.with_extension("tidy")).await; | ||
let held = held.unwrap(); | ||
mirror.tidy().await.unwrap(); | ||
drop(held); | ||
// A mirror with nowhere for the lock is not tidied. | ||
| @@ -219,11 +219,14 @@ mod tests { | ||
assert!(!waiter.is_finished()); | ||
drop(held); | ||
drop(waiter.await.unwrap().unwrap()); | ||
// Taken without waiting: there only when no one holds it. | ||
let held = Lock::try_acquire(path.clone()).unwrap().unwrap(); | ||
assert!(Lock::try_acquire(path.clone()).unwrap().is_none()); | ||
// Taken without waiting: there only when no one holds it. A lock let | ||
// go can stay held a moment longer, by a process another test forks | ||
// meanwhile until it runs git, so the free one is one no one has | ||
// opened. | ||
let fresh = dir.path().join("t.lock"); | ||
let held = Lock::try_acquire(fresh.clone()).unwrap().unwrap(); | ||
assert!(Lock::try_acquire(fresh.clone()).unwrap().is_none()); | ||
drop(held); | ||
assert!(Lock::try_acquire(path.clone()).unwrap().is_some()); | ||
assert!(Lock::try_acquire(dir.path().join("no/such/dir.lock")).is_err()); | ||
assert!( | ||
Lock::acquire(dir.path().join("no/such/dir.lock")) | ||
crates/knotless-serve/src/lib.rs+31−1
| @@ -80,6 +80,9 @@ pub struct App { | ||
pub reprove: std::sync::atomic::AtomicU64, | ||
/// Firehose events seen per DID. | ||
events: Mutex<HashMap<String, u64>>, | ||
/// The last commit, by its rev, that wrote to each DID's repository | ||
/// collections ([`first_of`]). | ||
revs: Mutex<HashMap<String, String>>, | ||
pub watching: AtomicBool, | ||
/// Syncs of each space that found something written there, by URI. | ||
moves: Mutex<HashMap<String, u64>>, | ||
| @@ -224,6 +227,7 @@ impl App { | ||
viewed: Mutex::default(), | ||
reprove: std::sync::atomic::AtomicU64::new(private::REPROVE.as_millis() as u64), | ||
events: Mutex::default(), | ||
revs: Mutex::default(), | ||
watching: AtomicBool::new(false), | ||
moves: Mutex::default(), | ||
following: AtomicBool::new(false), | ||
| @@ -364,7 +368,11 @@ impl App { | ||
.ops | ||
.iter() | ||
.any(|op| knotless::records::nsid::ALL.contains(&op.collection.as_str())); | ||
if event.ops.is_empty() || repository { | ||
// 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. | ||
let moved = event.ops.is_empty() || repository; | ||
if moved && first_of(&self.revs, &did, event.rev.as_deref()) { | ||
self.touch(&did); | ||
} | ||
if let Some(index) = self.index.clone() | ||
| @@ -639,6 +647,15 @@ pub(crate) fn aside(mirror: &Mirror) { | ||
tokio::spawn(async move { mirror.tidy().await.ok() }); | ||
} | ||
| ||
/// 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 { | ||
rev.is_none_or(|rev| { | ||
let last = seen.lock().unwrap().insert(did.into(), rev.into()); | ||
last.as_deref() != Some(rev) | ||
}) | ||
} | ||
| ||
/// 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. | ||
| @@ -1466,6 +1483,19 @@ fn mcp_appview(named: Option<String>, indexed: bool, addr: SocketAddr) -> Option | ||
| ||
#[cfg(test)] | ||
mod tests { | ||
| ||
#[test] | ||
fn one_commit_is_one_touch() { | ||
let seen = Mutex::default(); | ||
assert!(first_of(&seen, "did:a", Some("r1"))); | ||
// The rest of the commit's records, each an event of its own. | ||
assert!(!first_of(&seen, "did:a", Some("r1"))); | ||
assert!(!first_of(&seen, "did:a", Some("r1"))); | ||
assert!(first_of(&seen, "did:b", Some("r1"))); | ||
assert!(first_of(&seen, "did:a", Some("r2"))); | ||
assert!(first_of(&seen, "did:a", None)); | ||
assert!(first_of(&seen, "did:a", None)); | ||
} | ||
use super::*; | ||
| ||
#[test] | ||
docs/design/serve.md+6−0
| @@ -134,6 +134,12 @@ 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 | ||
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. | ||
| ||
With an index, the server also indexes each event. Records that arrive inline | ||
are written at once, in order, so a later write cannot land before an earlier | ||