| @@ -92,8 +92,7 @@ pub async fn follow(app: Arc<App>, every: Duration) { |
| | swept = Some(Instant::now()); |
| | } |
| | let due = std::mem::take(&mut *app.due.lock().unwrap()); |
| | let syncers = app.syncers().into_iter(); |
| | for (uri, did) in syncers.filter(|(uri, _)| all || due.contains(uri)) { |
| | for (uri, did) in to_sync(app.syncers(), all, &due) { |
| | let ctx = crate::login::syncer_ctx(&app, &did); |
| | // A credential asked for afresh now and then, which the owner's |
| | // PDS gives only to one still a member. |
| @@ -101,6 +100,7 @@ pub async fn follow(app: Arc<App>, every: Duration) { |
| | space::forget(&ctx); |
| | } |
| | let synced = sync(&ctx, &index, &uri).await; |
| | refused_for_scope(&did, &uri, &synced); |
| | let failed = synced.is_err(); |
| | if !failed { |
| | app.synced_now(&uri); |
| @@ -135,6 +135,22 @@ pub async fn follow(app: Arc<App>, every: Duration) { |
| | } |
| | } |
| | |
| | /// Of `syncers`, each a space and the member it is synced as, those to |
| | /// sync now: every one on a sweep (`all`), else those whose authority told |
| | /// of a write there (`due`). None whose member was refused it for want of |
| | /// scope lately ([`out_of_scope`]): their grant stays as it was until they |
| | /// log in again, and the space's readers sync it meanwhile, once it goes |
| | /// [`STALE`]. |
| | fn to_sync( |
| | syncers: Vec<(String, String)>, |
| | all: bool, |
| | due: &HashSet<String>, |
| | ) -> Vec<(String, String)> { |
| | let now = |(uri, _): &(String, String)| all || due.contains(uri); |
| | let asked = |(uri, did): &(String, String)| !out_of_scope(did, uri); |
| | syncers.into_iter().filter(now).filter(asked).collect() |
| | } |
| | |
| | /// Of [`space::denied`], what says the space itself is no more, rather |
| | /// than that its syncer may no longer read it. |
| | fn gone(err: &anyhow::Error) -> bool { |
| @@ -334,8 +350,9 @@ async fn owned(app: &App, index: &Index, did: &str) -> Result<()> { |
| | |
| | /// The spaces each account's session was refused for want of scope, by |
| | /// account and space, and when: not asked again until [`SCOPE_ASKED`] |
| | /// passes, as its grant stays as it was until it logs in again, and each |
| | /// asking costs a call to its PDS every [`DISCOVER`]. |
| | /// passes, or it logs in again ([`in_scope_again`]), as its grant stays as |
| | /// it was until then, and each asking costs a call to its PDS, every |
| | /// [`DISCOVER`] as a reader, every sweep of [`follow`] as a syncer. |
| | static OUT_OF_SCOPE: LazyLock<Mutex<HashMap<(String, String), Instant>>> = |
| | LazyLock::new(Mutex::default); |
| | |
| @@ -363,6 +380,12 @@ fn refused_for_scope<T>(did: &str, uri: &str, joined: &Result<T>) { |
| | } |
| | } |
| | |
| | /// Forgets the spaces `did`'s session was refused for want of scope: it |
| | /// was granted anew, as by logging in again, so each is asked again. |
| | pub fn in_scope_again(did: &str) { |
| | OUT_OF_SCOPE.lock().unwrap().retain(|(of, _), _| of != did); |
| | } |
| | |
| | /// Takes in what was written to the space `uri` since it was last synced, |
| | /// and says whether anything was. |
| | pub async fn sync(ctx: &Ctx, index: &Index, uri: &str) -> Result<bool> { |
| @@ -560,4 +583,24 @@ mod tests { |
| | OUT_OF_SCOPE.lock().unwrap().insert(key, long_ago); |
| | assert!(!out_of_scope(did, uri)); |
| | } |
| | |
| | #[test] |
| | fn syncers_out_of_scope_wait_till_they_log_in_again() { |
| | let did = "did:plc:syncer"; |
| | let (uri, other) = ( |
| | "at://did:plc:o/space/dev.knotless.space/r", |
| | "at://did:plc:o/space/dev.knotless.space/s", |
| | ); |
| | let syncers = || vec![(uri.into(), did.into()), (other.into(), did.into())]; |
| | let uris = |s: Vec<(String, String)>| s.into_iter().map(|(u, _)| u).collect::<Vec<_>>(); |
| | let none = HashSet::new(); |
| | assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]); |
| | let due = HashSet::from([other.to_string()]); |
| | assert_eq!(uris(to_sync(syncers(), false, &due)), [other]); |
| | let scope = anyhow::anyhow!("InvalidToken (403): Scope does not permit reading {uri}"); |
| | refused_for_scope::<()>(did, uri, &Err(scope)); |
| | assert_eq!(uris(to_sync(syncers(), true, &none)), [other]); |
| | in_scope_again(did); |
| | assert_eq!(uris(to_sync(syncers(), true, &none)), [uri, other]); |
| | } |
| | } |
| |
chadtmiller.com/knotless