| @@ -91,6 +91,12 @@ pub struct App { |
| | 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>>, |
| | /// Writes made from the pages (or by this server for an account), per |
| | /// DID: a read after one waits for its sync ([`App::served`]). |
| | writes: Mutex<HashMap<String, u64>>, |
| | /// Each public repository's mirror as last synced, by owner and name, |
| | /// for a read to take without waiting on the lock a sync holds. |
| | last: Mutex<HashMap<(String, String), Entry>>, |
| | /// Each repository's storage as last counted, kept until it moves |
| | /// (`dev.knotless.getRepoStorage`). |
| | pub(crate) counts: web::Counts, |
| @@ -180,7 +186,7 @@ pub struct App { |
| | |
| | type Shared<T> = Arc<tokio::sync::Mutex<T>>; |
| | |
| | #[derive(Default)] |
| | #[derive(Default, Clone)] |
| | struct Entry { |
| | git: Option<Git>, |
| | /// The repository's owner, key and name now, which differ from what |
| @@ -200,6 +206,27 @@ struct Entry { |
| | events: (u64, u64), |
| | /// Its space's moves when it was last synced, for a private one. |
| | moves: u64, |
| | /// The writes made from the pages for its owner when it was last |
| | /// synced ([`App::writes_for`]). |
| | writes: u64, |
| | /// When the first event came that made it stale, if one has since it |
| | /// was synced: kept with [`App::last`] only. |
| | stale_since: Option<Instant>, |
| | } |
| | |
| | /// How long a mirror that events have made stale is still served, to a |
| | /// read that wrote nothing, while its sync catches up. |
| | const STALE_FOR: Duration = Duration::from_secs(10); |
| | |
| | /// Whether a read is served `entry`, a mirror as it was last synced, |
| | /// without waiting on a sync, when its events now stamp as `stamp` and |
| | /// `wrote` writes were made from the pages: if no event has come since it |
| | /// synced, or the first came within [`STALE_FOR`], and the sync it started |
| | /// is then still catching up; never after a write from the pages that it |
| | /// may not have seen. |
| | fn serves(entry: &Entry, stamp: (u64, u64), wrote: u64) -> bool { |
| | let catching_up = entry.stale_since.is_some_and(|at| at.elapsed() < STALE_FOR); |
| | entry.writes == wrote && (entry.events == stamp || catching_up) |
| | } |
| | |
| | /// A repository found by name, and its mirror. |
| @@ -258,6 +285,8 @@ impl App { |
| | events: Mutex::default(), |
| | account_events: Mutex::default(), |
| | repo_events: Mutex::default(), |
| | writes: Mutex::default(), |
| | last: Mutex::default(), |
| | counts: web::Counts::default(), |
| | revs: Mutex::default(), |
| | watching: AtomicBool::new(false), |
| @@ -420,7 +449,7 @@ impl App { |
| | 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()); |
| | self.heard_of(&did, key.as_deref(), false); |
| | } |
| | } |
| | if let Some(index) = self.index.clone() |
| @@ -556,19 +585,61 @@ 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); |
| | self.heard_of(did, None, true); |
| | } |
| | |
| | /// The writes from the pages recorded for `did` ([`App::touch`]). |
| | fn writes_for(&self, did: &str) -> u64 { |
| | self.writes.lock().unwrap().get(did).copied().unwrap_or(0) |
| | } |
| | |
| | /// The mirror of `did`'s repository `name` as last synced, if a read |
| | /// may have it now ([`serves_last`]): never unless the firehose is |
| | /// watching, which is what makes the sync of a stale one certain. |
| | fn served(&self, did: &str, name: &str, wrote: u64) -> Option<Entry> { |
| | if !self.watching.load(Ordering::Relaxed) { |
| | return None; |
| | } |
| | let last = self.last.lock().unwrap(); |
| | let entry = last.get(&(did.to_string(), name.to_string()))?; |
| | let stamp = self.stamp(did, entry.key(did).as_deref()); |
| | serves(entry, stamp, wrote).then(|| entry.clone()) |
| | } |
| | |
| | /// Keeps `entry`, just synced, as what [`App::served`] gives, or none |
| | /// if the repository is not there. |
| | fn remember(&self, did: &str, name: &str, entry: &Entry) { |
| | let mut last = self.last.lock().unwrap(); |
| | let at = (did.to_string(), name.to_string()); |
| | match entry.git { |
| | Some(_) => last.insert(at, entry.clone()), |
| | None => last.remove(&at), |
| | }; |
| | } |
| | |
| | /// 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>) { |
| | fn heard_of(self: &Arc<App>, did: &str, key: Option<&str>, own: bool) { |
| | *self |
| | .events |
| | .lock() |
| | .unwrap() |
| | .entry(did.to_string()) |
| | .or_default() += 1; |
| | let now = Instant::now(); |
| | for (_, entry) in self |
| | .last |
| | .lock() |
| | .unwrap() |
| | .iter_mut() |
| | .filter(|(at, _)| at.0 == did) |
| | { |
| | entry.stale_since.get_or_insert(now); |
| | } |
| | if own { |
| | *self.writes.lock().unwrap().entry(did.into()).or_default() += 1; |
| | } |
| | match key { |
| | Some(key) => { |
| | let mut repos = self.repo_events.lock().unwrap(); |
| @@ -595,7 +666,7 @@ impl App { |
| | let app = self.clone(); |
| | let did = did.to_string(); |
| | tokio::spawn(async move { |
| | let _ = app.mirror(&did, &name).await; |
| | let _ = app.located_with(&did, &name, true).await; |
| | }); |
| | } |
| | } |
| @@ -609,6 +680,29 @@ impl App { |
| | /// name it has now: a public repository, or a private one the viewer |
| | /// of the request being answered may read ([`private`]). |
| | pub async fn located(&self, authority: &str, name: &str) -> Result<Located, AppError> { |
| | self.located_with(authority, name, true).await |
| | } |
| | |
| | /// [`App::located`] for a read of the repository that answers a request: |
| | /// a mirror a change has made stale may be served while it syncs |
| | /// ([`App::served`]). Nothing that derives and keeps something from the |
| | /// mirror may read it so. |
| | pub async fn located_for_reader( |
| | &self, |
| | authority: &str, |
| | name: &str, |
| | ) -> Result<Located, AppError> { |
| | self.located_with(authority, name, false).await |
| | } |
| | |
| | /// [`App::located`], waiting for the sync of a stale mirror when `wait`, |
| | /// as the refresh a change starts does: it is the sync. |
| | async fn located_with( |
| | &self, |
| | authority: &str, |
| | name: &str, |
| | wait: bool, |
| | ) -> Result<Located, AppError> { |
| | let url = RepoUrl::parse(&format!("knotless://{authority}/{name}")) |
| | .map_err(|e| AppError(StatusCode::BAD_REQUEST, e.to_string()))?; |
| | let remote = Remote::connect(&self.ctx, &url).await; |
| @@ -635,7 +729,8 @@ impl App { |
| | return read; |
| | } |
| | } |
| | let missing = match self.public(remote, &url).await { |
| | let owner = wait || viewer.as_deref() == Some(did.as_str()); |
| | let missing = match self.public(remote, &url, owner).await { |
| | Err(e) if e.0 == StatusCode::NOT_FOUND => e, |
| | found => return found, |
| | }; |
| @@ -649,8 +744,22 @@ impl App { |
| | } |
| | |
| | /// [`App::located`] for a repository anyone may read. |
| | async fn public(&self, remote: Remote, url: &RepoUrl) -> Result<Located, AppError> { |
| | async fn public( |
| | &self, |
| | remote: Remote, |
| | url: &RepoUrl, |
| | owner: bool, |
| | ) -> Result<Located, AppError> { |
| | let did = remote.did().to_string(); |
| | // Watched, a mirror events have made stale is still served for a |
| | // moment, while the sync they started finishes: a read does not |
| | // wait on the lock that sync holds, unless its owner made it, who |
| | // may have just written with a tool of their own. |
| | let wrote = self.writes_for(&did); |
| | let view = self.served(&did, &url.name, wrote).filter(|_| !owner); |
| | if let Some(view) = view { |
| | return self.admit(url, self.found(&view, &did, self.ctx.clone())); |
| | } |
| | let entry = self |
| | .repos |
| | .lock() |
| @@ -682,9 +791,17 @@ impl App { |
| | let found = entry.key(&did); |
| | entry.events = self.settled(&did, key.as_deref(), events, any, found.as_deref()); |
| | } |
| | if synced.is_ok() { |
| | entry.writes = wrote; |
| | } |
| | self.remember(&did, &url.name, &entry); |
| | synced?; |
| | } |
| | let found = self.found(&entry, &did, self.ctx.clone()); |
| | self.admit(url, self.found(&entry, &did, self.ctx.clone())) |
| | } |
| | |
| | /// `found`, unless it is taken down. |
| | fn admit(&self, url: &RepoUrl, found: Located) -> Result<Located, AppError> { |
| | let uri = knotless::forge::repo_uri(&found.did, &found.name); |
| | if let Some(index) = &self.index |
| | && index |
| @@ -1753,6 +1870,28 @@ mod tests { |
| | assert_eq!(scopes(&[]), [None]); |
| | } |
| | |
| | #[test] |
| | fn a_stale_mirror_is_served_while_it_syncs() { |
| | let long_ago = Instant::now().checked_sub(STALE_FOR * 2).unwrap(); |
| | let mut entry = Entry { |
| | events: (1, 1), |
| | ..Entry::default() |
| | }; |
| | // Quiet, it is served however long ago it synced. |
| | entry.synced = Some(long_ago); |
| | assert!(serves(&entry, (1, 1), 0)); |
| | // Stale with no event noted, or the first long ago, it is not. |
| | assert!(!serves(&entry, (2, 1), 0)); |
| | entry.stale_since = Some(long_ago); |
| | assert!(!serves(&entry, (2, 1), 0)); |
| | // The first event only now, it is, however long ago it synced. |
| | entry.stale_since = Some(Instant::now()); |
| | assert!(serves(&entry, (2, 1), 0)); |
| | // A write from the pages it may not have seen waits for the sync. |
| | assert!(!serves(&entry, (2, 1), 1)); |
| | assert!(!serves(&entry, (1, 1), 1)); |
| | } |
| | |
| | #[test] |
| | fn one_commit_is_one_touch() { |
| | let seen = Mutex::default(); |
| |
chadtmiller.com/knotless