chadtmiller.com/knotless
git that lives in your Atmosphere account
1
10
A private repository's readers sync it once its syncer has not lately
crates/knotless-serve/src/lib.rs+19−0
| @@ -127,6 +127,11 @@ pub struct App { | ||
due: Mutex<std::collections::HashSet<String>>, | ||
/// Wakes [`spaces::follow`] for what is due. | ||
told: tokio::sync::Notify, | ||
/// When each space was last synced in full, as anyone here synced it: | ||
/// a read syncs it itself once that is [`spaces::STALE`] ago, as when | ||
/// the session of the member it is synced as no longer reads it. Its | ||
/// entries taken out, every space counts as not synced lately. | ||
pub last_synced: Mutex<HashMap<String, Instant>>, | ||
/// When this server last asked each space's authority to tell it of | ||
/// writes there. | ||
registered: tokio::sync::Mutex<HashMap<String, Instant>>, | ||
| @@ -254,6 +259,7 @@ impl App { | ||
with_spaces: Mutex::default(), | ||
due: Mutex::default(), | ||
told: tokio::sync::Notify::new(), | ||
last_synced: Mutex::default(), | ||
registered: tokio::sync::Mutex::default(), | ||
code: None, | ||
usage: None, | ||
| @@ -487,6 +493,19 @@ impl App { | ||
self.moves.lock().unwrap().get(uri).copied().unwrap_or(0) | ||
} | ||
| ||
/// Notes that the space `uri` was synced just now. | ||
pub(crate) fn synced_now(&self, uri: &str) { | ||
let mut last = self.last_synced.lock().unwrap(); | ||
last.insert(uri.to_string(), Instant::now()); | ||
} | ||
| ||
/// Whether the space `uri` was not synced within [`spaces::STALE`], | ||
/// or not since this server started. | ||
pub(crate) fn sync_stale(&self, uri: &str) -> bool { | ||
let last = self.last_synced.lock().unwrap(); | ||
last.get(uri).is_none_or(|at| at.elapsed() >= spaces::STALE) | ||
} | ||
| ||
/// Records that the space `uri` was written to, or is no longer synced, | ||
/// so that mirrors of it sync again as they are next read. | ||
pub fn moved(&self, uri: &str) { | ||
crates/knotless-serve/src/private.rs+40−6
| @@ -271,10 +271,42 @@ impl App { | ||
kept.map_err(AppError::internal)?; | ||
let synced = index.space_synced(uri).map_err(anyhow::Error::from); | ||
let synced = synced.map_err(AppError::internal)?; | ||
let sync = !synced || self.syncer_of(uri).is_none(); | ||
let sync = sync.then(|| crate::spaces::sync_with(ctx, index, joined)); | ||
let read = futures::future::OptionFuture::from(sync).await.transpose(); | ||
read.map(drop).map_err(AppError::upstream) | ||
// Synced as the reader, too, when its syncer has not lately. | ||
let sync = !synced || self.syncer_of(uri).is_none() || self.sync_stale(uri); | ||
if !sync { | ||
return Ok(()); | ||
} | ||
let read = crate::spaces::sync_with(ctx, index, joined).await; | ||
self.read_now(uri, read, synced) | ||
} | ||
| ||
/// What a reader's sync of the space `uri` makes of their read, with | ||
/// `synced` whether it was ever synced before. A sync that failed | ||
/// counts as one made, so it is tried again only after | ||
/// [`STALE`](crate::spaces::STALE), not at each read; and a space | ||
/// already synced is then read from its index as it was, the failure | ||
/// logged (`space not synced`), rather than failing the read. One never | ||
/// synced has nothing to be read from, and its failure is the read's. | ||
fn read_now( | ||
&self, | ||
uri: &str, | ||
read: anyhow::Result<bool>, | ||
synced: bool, | ||
) -> Result<(), AppError> { | ||
if read.is_ok() || synced { | ||
self.synced_now(uri); | ||
} | ||
if matches!(read, Ok(true)) { | ||
self.moved(uri); | ||
} | ||
let read = read.map(drop); | ||
let (logged, read) = if synced { | ||
(read, Ok(())) | ||
} else { | ||
(Ok(()), read) | ||
}; | ||
crate::telemetry::space_synced(uri, logged); | ||
read.map_err(AppError::upstream) | ||
} | ||
| ||
/// Notes whether `viewer` may read the private repository in the | ||
| @@ -309,12 +341,14 @@ impl App { | ||
self.follow_org(space, syncer); | ||
let synced = index.space_synced(space); | ||
let synced = synced.map_err(anyhow::Error::from); | ||
if synced.map_err(AppError::internal)? { | ||
let synced = synced.map_err(AppError::internal)?; | ||
// Synced, and lately: what was written since its sync will reach it. | ||
if synced && !self.sync_stale(space) { | ||
return Ok(()); | ||
} | ||
let ctx = crate::login::syncer_ctx(self, syncer); | ||
let read = crate::spaces::sync(&ctx, index, space).await; | ||
read.map(drop).map_err(AppError::upstream) | ||
self.read_now(space, read, synced) | ||
} | ||
| ||
/// Stops syncing the space `uri`, and forgets what it gave: its sync | ||
crates/knotless-serve/src/spaces.rs+10−0
| @@ -66,6 +66,13 @@ const COLLECTIONS: [&str; 18] = [ | ||
nsid::ORG_ACCEPTANCE, | ||
]; | ||
| ||
/// How long a space may go unsynced before a member reading it syncs it | ||
/// with their own credential: two of [`follow`]'s sweeps. A space's | ||
/// syncer that fails, as one whose session here lapsed or lacks the | ||
/// scope its space needs does, would otherwise leave it as it was until | ||
/// they log in again, every write since missing from it, to every reader. | ||
pub const STALE: Duration = Duration::from_secs(60); | ||
| ||
/// Syncs each space `app` keeps an index of as its authority tells of | ||
/// writes there, and every one of them every `every` besides, for as long | ||
/// as it runs, each as the member it is synced with. | ||
| @@ -91,6 +98,9 @@ pub async fn follow(app: Arc<App>, every: Duration) { | ||
} | ||
let synced = sync(&ctx, &index, &uri).await; | ||
let failed = synced.is_err(); | ||
if !failed { | ||
app.synced_now(&uri); | ||
} | ||
// Refused, as once the space is deleted or its syncer no longer | ||
// a member, it is followed no more; its authority not reached, | ||
// as while it restarts, it is synced again at the next. | ||
crates/knotless-serve/tests/gateway.rs+116−0
| @@ -3450,6 +3450,122 @@ fn private_repositories_heard_of_from_clients() { | ||
assert!(heard(&member).is_empty()); | ||
} | ||
| ||
/// A private repository's index stays current when the member it is | ||
/// synced as can no longer sync it, as when their session here lapses or | ||
/// lacks the scope its space needs: a member reading it syncs it with | ||
/// their own credential once it was not synced within a sweep or two, | ||
/// rather than every reader seeing it as it was until that member logs in | ||
/// again. A reader's sync that fails leaves the read to the index as it | ||
/// was, and is not tried again at the next read. | ||
#[test] | ||
fn private_repositories_synced_by_readers_once_their_syncer_lapses() { | ||
if !net().spaces { | ||
return; | ||
} | ||
let server = gateway_with(None, true); | ||
let [mut owner, mut member, mut other] = people(); | ||
for s in [&mut owner, &mut member, &mut other] { | ||
s.set("KNOTLESS_APPVIEW", &server.base); | ||
} | ||
let run = |s: &Sandbox, args: &[&str]| { | ||
let (code, out) = s.cli(args, b""); | ||
assert_eq!(code, 0, "{args:?}: {out}"); | ||
out.trim().to_string() | ||
}; | ||
let url = owner.url("ledger"); | ||
run(&owner, &["create", &url, "--private"]); | ||
let work = owner.work("w", 1); | ||
owner.git(&work, &["push", "-q", &url, "main"]); | ||
run(&owner, &["share", &url, &member.account.handle]); | ||
run(&owner, &["share", &url, &other.account.handle]); | ||
let did = owner.account.did.clone(); | ||
let at = format!("at://{did}/dev.knotless.repo/ledger"); | ||
let client = reqwest::blocking::Client::new(); | ||
| ||
// The owner reads it on the pages, and so it is synced as them. | ||
let cookie = session_cookie(&server, &owner.account); | ||
let res = client | ||
.get(format!( | ||
"{}/xrpc/dev.knotless.getRepo?repo={at}", | ||
server.base | ||
)) | ||
.header("cookie", &cookie) | ||
.send() | ||
.unwrap(); | ||
assert_eq!(res.status().as_u16(), 200); | ||
let syncers = server.app.syncers(); | ||
assert!(syncers.iter().any(|(_, d)| *d == did), "{syncers:?}"); | ||
| ||
// Their session here lapses, and they stay its syncer: each sync as | ||
// them fails, though not as one refused, which would let it go. | ||
let pages = knotless_serve::login::ctx_for(&server.app, &did); | ||
knotless::session::remove(&pages, &did); | ||
run(&member, &["issue", &url, "Written after the lapse"]); | ||
| ||
// The issues a member reads, with a delegation token of theirs, each | ||
// spent once; once the space counts as not synced lately. | ||
let rt = tokio::runtime::Runtime::new().unwrap(); | ||
let c = ctx(&member); | ||
let me = rt.block_on(knotless::session::me(&c)).unwrap(); | ||
let space = knotless::space::uri(&did, "ledger"); | ||
let stale = || server.app.last_synced.lock().unwrap().clear(); | ||
let listed = || { | ||
let token = rt.block_on(knotless::space::delegation(&c, &space, &me)); | ||
let res = client | ||
.get(format!( | ||
"{}/xrpc/dev.knotless.listItems?repo={at}&kind=issue", | ||
server.base | ||
)) | ||
.header(knotless::space::DELEGATION_HEADER, token.unwrap()) | ||
.send() | ||
.unwrap(); | ||
assert_eq!(res.status().as_u16(), 200); | ||
let listed: Value = res.json().unwrap(); | ||
let mut titles: Vec<String> = listed["items"] | ||
.as_array() | ||
.unwrap() | ||
.iter() | ||
.filter_map(|i| i["title"].as_str().map(String::from)) | ||
.collect(); | ||
titles.sort(); | ||
titles | ||
}; | ||
| ||
// Once it was not synced lately, a member's read syncs it as them. | ||
stale(); | ||
assert_eq!(listed(), ["Written after the lapse"]); | ||
| ||
// A member whose repo cannot be read fails the sync: one who writes, | ||
// then is taken down where they live. The read is answered from the | ||
// index as it was, and the sync that failed counts as one made. | ||
run(&other, &["issue", &url, "Written before the takedown"]); | ||
let takedown = |applied: bool| { | ||
let subject = json!({"$type": "com.atproto.admin.defs#repoRef", "did": other.account.did}); | ||
let res = client | ||
.post(format!( | ||
"{}/xrpc/com.atproto.admin.updateSubjectStatus", | ||
net().pds | ||
)) | ||
.header("authorization", format!("Basic {}", admin())) | ||
.json(&json!({"subject": subject, "takedown": {"applied": applied}})) | ||
.send() | ||
.unwrap(); | ||
assert!(res.status().is_success(), "{}", res.text().unwrap()); | ||
}; | ||
takedown(true); | ||
stale(); | ||
assert_eq!(listed(), ["Written after the lapse"]); | ||
assert!(server.app.last_synced.lock().unwrap().contains_key(&space)); | ||
| ||
// Restored, it is taken in at the next read that finds it stale. | ||
takedown(false); | ||
stale(); | ||
assert_eq!( | ||
listed(), | ||
["Written after the lapse", "Written before the takedown"] | ||
); | ||
} | ||
| ||
/// From the repository's settings, its owner sees its runners' keys, and | ||
/// leaves, lists and takes back secrets sealed to them in the browser, | ||
/// with the pages' login rather than a service token. | ||
docs/design/access.md+12−1
| @@ -251,7 +251,8 @@ sequenceDiagram | ||
and keeps the credential under their DID for what reads the space as | ||
them after. It proves them a member of that repository alone. With no | ||
session here to sync its space with after, the space is synced with | ||
that credential at each such query, when no member's session syncs it. | ||
that credential at each such query, when no member's session syncs it, | ||
or when that session has not synced it within a minute. | ||
- Anyone else, logged in or not, is answered 404 with the words a | ||
repository never made gets, so a private repository's existence does not | ||
show. | ||
| @@ -291,6 +292,16 @@ repository on the pages is a repository in that index. | ||
`did:web` holds no repo, and the alpha lets no one but the owner list a | ||
space's members. Which member each space is synced as is kept in the | ||
index, so syncing goes on across restarts. | ||
- A space not synced within a minute, two of the sweeps that sync every | ||
space every 30 seconds, or not since the server started, is synced by | ||
whoever reads it next, with their own session or delegation token, | ||
before the page is answered. Its syncer's sync can fail without its | ||
being refused, as when their session here lapses or its grant leaves | ||
the space out; until they logged in again, every write since would | ||
otherwise be missing from it, for every reader. A reader's sync that | ||
fails is logged, `space not synced`, and the repository is answered from | ||
the index as it was, then tried again a minute on; only one never synced | ||
fails the read. | ||
- One made with `knotless create --private`, the `create_repo` tool or a | ||
fork is in the index before anyone looks, if its owner is logged in to | ||
the appview the command names (`KNOTLESS_APPVIEW`): the command asks the | ||