alice.pds.demo.boltless.dev
This repository has no description
15 kB
442 lines
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::sync::atomic::{AtomicUsize, Ordering};
2use std::sync::{Arc, Mutex};
3
4use futures::StreamExt;
5use knot_atproto::Atproto;
6use knot_cob::{CobHome, CobStore};
7use knot_cobs::{Grant, MembersChange, Registration, RegistryChange};
8use knot_git::{Layout, Repo};
9use knot_index::{Coverage, Index, KeyReprieve, KeyTtl, Resolved, SweepFloor};
10use knot_keyfill::{
11 AccountBudget, BusyRetry, Cursors, Pace, SettleFloor, SettledPause, StalledBackoff, fill_once,
12};
13use knot_resource::{Burst, HostKey, HostPacer, RateLimit, RefillMicros, Slots};
14use knot_runtime::{
15 FakeHttp, HttpRequest, HttpResponse, K256Signer, ManualClock, NetworkError, SeededEntropy,
16 Signer, UnixMicros,
17};
18use knot_types::{
19 AccountDid, KnotId, OfferedKey, OwnerDid, RepoDid, RepoName, RepoRkey, UnixSeconds,
20};
21use tempfile::TempDir;
22use tokio_util::sync::CancellationToken;
23use url::Url;
24
25const KNOT_DID: &str = "did:web:nel.pet";
26const NEL: &str = "did:plc:nel";
27const OLAREN: &str = "did:plc:olaren";
28const TEQ: &str = "did:plc:teq";
29const BAILEY: &str = "did:plc:bailey";
30const NEL_PDS: &str = "https://pds.nel.pet";
31const OLAREN_PDS: &str = "https://pds.olaren.dev";
32
33type Responder = Box<dyn Fn(&HttpRequest) -> Result<HttpResponse, NetworkError> + Send + Sync>;
34
35fn responding(status: http::StatusCode, body: bytes::Bytes) -> HttpResponse {
36 HttpResponse {
37 status,
38 headers: http::HeaderMap::new(),
39 body,
40 }
41}
42
43struct Hosted {
44 owner: OwnerDid,
45 repo: RepoDid,
46 name: RepoName,
47}
48
49fn account(did: &str) -> AccountDid {
50 AccountDid::new(did).unwrap()
51}
52
53fn anemone() -> Hosted {
54 Hosted {
55 owner: OwnerDid::new(NEL).unwrap(),
56 repo: RepoDid::new("did:plc:squid").unwrap(),
57 name: RepoName::new("anemone").unwrap(),
58 }
59}
60
61fn barnacle() -> Hosted {
62 Hosted {
63 owner: OwnerDid::new(OLAREN).unwrap(),
64 repo: RepoDid::new("did:plc:limpet").unwrap(),
65 name: RepoName::new("barnacle").unwrap(),
66 }
67}
68
69fn hosted_index(scratch: &TempDir, repos: &[Hosted], members: &[AccountDid]) -> Arc<Index> {
70 let meta_path = scratch.path().join("meta");
71 Repo::create(&meta_path).unwrap();
72 let layout = Layout::new(scratch.path().join("repos"));
73 let meta = Repo::open(&meta_path).unwrap();
74 let store = CobStore::new(&meta);
75 let home = CobHome::from(&KnotId::new(KNOT_DID).unwrap());
76 let signer = K256Signer::generate(&SeededEntropy::new(3));
77 let registration = |hosted: &Hosted| {
78 layout.create(&hosted.repo).unwrap();
79 RegistryChange::Register(Registration {
80 owner: hosted.owner.clone(),
81 rkey: RepoRkey::new(hosted.name.as_str()).unwrap(),
82 name: hosted.name.clone(),
83 repo: hosted.repo.clone(),
84 created_at: UnixSeconds::new(1),
85 })
86 };
87 let (first, rest) = repos.split_first().expect("a knot under test hosts a repo");
88 let registry = store
89 .create(&home, ®istration(first), &signer, UnixSeconds::new(1))
90 .unwrap()
91 .object;
92 rest.iter().for_each(|hosted| {
93 store
94 .update(
95 &home,
96 registry,
97 ®istration(hosted),
98 &signer,
99 UnixSeconds::new(2),
100 )
101 .unwrap();
102 });
103
104 let granted = |subject: &AccountDid| {
105 MembersChange::Add(Grant {
106 subject: subject.clone(),
107 added_by: account(NEL),
108 created_at: UnixSeconds::new(1),
109 })
110 };
111 if let Some((first, rest)) = members.split_first() {
112 let roll = store
113 .create(&home, &granted(first), &signer, UnixSeconds::new(1))
114 .unwrap()
115 .object;
116 rest.iter().for_each(|subject| {
117 store
118 .update(&home, roll, &granted(subject), &signer, UnixSeconds::new(2))
119 .unwrap();
120 });
121 }
122
123 let index = Arc::new(Index::new(meta_path, layout));
124 index.rebuild().unwrap();
125 index.warm_collaborators();
126 index
127}
128
129fn atproto_with(responder: Responder) -> Arc<Atproto<FakeHttp<Responder>, ManualClock>> {
130 Arc::new(Atproto::new(
131 FakeHttp::new(responder),
132 ManualClock::new(UnixMicros::new(1_000_000_000)),
133 KnotId::new(KNOT_DID).unwrap(),
134 knot_atproto::PlcDirectory::new(Url::parse("https://plc.directory/").unwrap()).unwrap(),
135 ))
136}
137
138fn atproto_answering(
139 status: http::StatusCode,
140 calls: Arc<AtomicUsize>,
141) -> Arc<Atproto<FakeHttp<Responder>, ManualClock>> {
142 atproto_with(Box::new(move |_request: &HttpRequest| {
143 calls.fetch_add(1, Ordering::SeqCst);
144 Ok(responding(status, bytes::Bytes::new()))
145 }))
146}
147
148fn did_document(did: &str, handle: &str, pds: &str) -> bytes::Bytes {
149 let signing = K256Signer::generate(&SeededEntropy::new(5));
150 let multibase = knot_types::crypto::multikey(0xe7, signing.public_key().as_bytes());
151 let body = serde_json::json!({
152 "id": did,
153 "alsoKnownAs": [format!("at://{handle}")],
154 "verificationMethod": [{
155 "id": format!("{did}#atproto"),
156 "type": "Multikey",
157 "controller": did,
158 "publicKeyMultibase": multibase,
159 }],
160 "service": [{
161 "id": "#atproto_pds",
162 "type": "AtprotoPersonalDataServer",
163 "serviceEndpoint": pds,
164 }]
165 });
166 bytes::Bytes::from(serde_json::to_vec(&body).unwrap())
167}
168
169fn atproto_serving_two_accounts() -> Arc<Atproto<FakeHttp<Responder>, ManualClock>> {
170 atproto_with(Box::new(move |request: &HttpRequest| {
171 let url = request.url.as_str();
172 let body = if url.contains("listRecords") {
173 bytes::Bytes::from_static(br#"{"records":[]}"#)
174 } else if url.ends_with(NEL) {
175 did_document(NEL, "nel.pet", NEL_PDS)
176 } else if url.ends_with(OLAREN) {
177 did_document(OLAREN, "olaren.dev", OLAREN_PDS)
178 } else {
179 return Ok(responding(http::StatusCode::NOT_FOUND, bytes::Bytes::new()));
180 };
181 Ok(responding(http::StatusCode::OK, body))
182 }))
183}
184
185fn atproto_recording(
186 seen: Arc<Mutex<Vec<String>>>,
187) -> Arc<Atproto<FakeHttp<Responder>, ManualClock>> {
188 atproto_with(Box::new(move |request: &HttpRequest| {
189 let url = request.url.as_str();
190 if url.contains("listRecords") {
191 return Ok(responding(
192 http::StatusCode::OK,
193 bytes::Bytes::from_static(br#"{"records":[]}"#),
194 ));
195 }
196 if url.ends_with(NEL) {
197 return Ok(responding(
198 http::StatusCode::OK,
199 did_document(NEL, "nel.pet", NEL_PDS),
200 ));
201 }
202 if let Some(did) = url.rsplit('/').next() {
203 seen.lock().unwrap().push(did.to_string());
204 }
205 Ok(responding(
206 http::StatusCode::SERVICE_UNAVAILABLE,
207 bytes::Bytes::new(),
208 ))
209 }))
210}
211
212fn now() -> UnixSeconds {
213 UnixSeconds::new(1_000)
214}
215
216fn brisk() -> Pace {
217 Pace {
218 busy: BusyRetry::from_millis(0),
219 floor: SettleFloor::from_millis(0),
220 ttl: KeyTtl::from_secs(3_600),
221 reprieve: KeyReprieve::from_secs(300, 21_600),
222 sweep: SweepFloor::DEFAULT,
223 stalled: StalledBackoff::from_secs(1),
224 settled: SettledPause::from_secs(1),
225 members: AccountBudget::new(64),
226 suspected: AccountBudget::new(256),
227 host: RateLimit {
228 burst: Burst::new(1),
229 refill: RefillMicros::new(1_000),
230 },
231 }
232}
233
234#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
235async fn a_gone_document_completes_the_set_and_an_outage_leaves_it_warming() {
236 let outcome = |status: http::StatusCode| async move {
237 let scratch = tempfile::tempdir().unwrap();
238 let index = hosted_index(&scratch, &[anemone()], &[]);
239 let calls = Arc::new(AtomicUsize::new(0));
240 let atproto = atproto_answering(status, Arc::clone(&calls));
241 let pacer = HostPacer::new(brisk().host);
242 assert_eq!(index.keys().coverage(), Coverage::Warming);
243 fill_once(
244 &index,
245 &atproto,
246 &Slots::testing(4),
247 &pacer,
248 brisk(),
249 &Cursors::default(),
250 )
251 .await;
252 assert!(
253 calls.load(Ordering::SeqCst) > 0,
254 "the fill made an outbound request"
255 );
256 index.keys().coverage()
257 };
258
259 assert_eq!(
260 outcome(http::StatusCode::NOT_FOUND).await,
261 Coverage::Ready,
262 "a permanently unresolvable owner is recorded with an empty key set, so the set is complete"
263 );
264 assert_eq!(
265 outcome(http::StatusCode::SERVICE_UNAVAILABLE).await,
266 Coverage::Warming,
267 "a transient failure doesn't teach the knot anything, so it mustn't claim a complete set"
268 );
269}
270
271#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
272async fn keys_the_knot_already_read_survive_an_outage_and_a_refused_listing() {
273 let scratch = tempfile::tempdir().unwrap();
274 let index = hosted_index(&scratch, &[anemone()], &[]);
275 let key = OfferedKey::from_bytes(vec![9]);
276 let spent = KeyTtl::from_secs(1).lease_from(UnixSeconds::new(100));
277 index.keys().record(&account(NEL), vec![key.clone()], spent);
278 let pacer = HostPacer::new(brisk().host);
279
280 let unreachable = atproto_answering(http::StatusCode::SERVICE_UNAVAILABLE, Arc::default());
281 fill_once(
282 &index,
283 &unreachable,
284 &Slots::testing(4),
285 &pacer,
286 brisk(),
287 &Cursors::default(),
288 )
289 .await;
290 assert_eq!(
291 index.keys().coverage(),
292 Coverage::Ready,
293 "an owner the knot has read before keeps its last keys through an outage, so one \
294 unreachable PDS mustn't reopen the knot to every offered key"
295 );
296 assert_eq!(
297 index.owner_of_key(&key, now()),
298 Resolved::Ready(Some(account(NEL))),
299 "the reprieve keeps the keys the knot last read"
300 );
301
302 index.keys().record(&account(NEL), vec![key.clone()], spent);
303 let listing = Arc::new(AtomicUsize::new(0));
304 let refusing = {
305 let listing = Arc::clone(&listing);
306 atproto_with(Box::new(move |request: &HttpRequest| {
307 let url = request.url.as_str();
308 if url.contains("listRecords") {
309 listing.fetch_add(1, Ordering::SeqCst);
310 return Ok(responding(
311 http::StatusCode::BAD_REQUEST,
312 bytes::Bytes::new(),
313 ));
314 }
315 Ok(responding(
316 http::StatusCode::OK,
317 did_document(NEL, "nel.pet", NEL_PDS),
318 ))
319 }))
320 };
321 fill_once(
322 &index,
323 &refusing,
324 &Slots::testing(4),
325 &pacer,
326 brisk(),
327 &Cursors::default(),
328 )
329 .await;
330 assert!(
331 listing.load(Ordering::SeqCst) > 0,
332 "the fill read from the PDS"
333 );
334 assert_eq!(
335 index.owner_of_key(&key, now()),
336 Resolved::Ready(Some(account(NEL))),
337 "an account whose DID document resolves hasn't gone anywhere, so the knot mustn't read \
338 a 400 from a record listing as proof the account stopped publishing keys"
339 );
340}
341
342#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
343async fn the_fill_takes_a_turn_at_each_host_it_reads_from_and_stops_on_shutdown() {
344 let scratch = tempfile::tempdir().unwrap();
345 let index = hosted_index(&scratch, &[anemone(), barnacle()], &[]);
346 let atproto = atproto_serving_two_accounts();
347 let pacer = HostPacer::new(brisk().host);
348
349 fill_once(
350 &index,
351 &atproto,
352 &Slots::testing(4),
353 &pacer,
354 brisk(),
355 &Cursors::default(),
356 )
357 .await;
358
359 assert_eq!(
360 index.keys().coverage(),
361 Coverage::Ready,
362 "both owners resolved, so every account that may push has a record"
363 );
364 ["plc.directory", "pds.nel.pet", "pds.olaren.dev"]
365 .iter()
366 .for_each(|host| {
367 assert!(
368 !pacer.reserve_now(&HostKey::new(host), UnixMicros::new(0)),
369 "the fill must take a turn at {host} before it reads from it, \
370 or a knot whose members share one PDS spends its whole rate at that host"
371 );
372 });
373 assert!(
374 pacer.reserve_now(&HostKey::new("pds.teq.dev"), UnixMicros::new(0)),
375 "a host the fill never read from is due immediately. The bookings above are the fill's \
376 own work"
377 );
378
379 let shutdown = CancellationToken::new();
380 let task = knot_keyfill::spawn(
381 Arc::clone(&index),
382 Arc::clone(&atproto),
383 Slots::testing(4),
384 brisk(),
385 shutdown.clone(),
386 );
387 shutdown.cancel();
388 tokio::time::timeout(std::time::Duration::from_secs(5), task)
389 .await
390 .expect("a shutting-down knot mustn't wait out the pause between fill passes")
391 .expect("the fill task stops without panicking");
392}
393
394#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
395async fn members_are_read_whole_on_first_contact_then_renewed_one_budget_turn_per_pass() {
396 let scratch = tempfile::tempdir().unwrap();
397 let members = [account(OLAREN), account(TEQ), account(BAILEY)];
398 let index = hosted_index(&scratch, &[anemone()], &members);
399 let seen = Arc::new(Mutex::new(Vec::new()));
400 let atproto = atproto_recording(Arc::clone(&seen));
401 let pacer = HostPacer::new(brisk().host);
402 let pace = Pace {
403 members: AccountBudget::new(1),
404 ..brisk()
405 };
406 let cursor = Cursors::default();
407
408 fill_once(&index, &atproto, &Slots::testing(4), &pacer, pace, &cursor).await;
409 assert_eq!(
410 index.keys().coverage(),
411 Coverage::Ready,
412 "the pushers are what coverage waits on, so an unreadable member mustn't make the knot \
413 doubt the keys it checks pushes against"
414 );
415 let mut attempted = seen.lock().unwrap().clone();
416 attempted.sort();
417 assert_eq!(
418 attempted,
419 vec![BAILEY.to_string(), OLAREN.to_string(), TEQ.to_string()],
420 "the renewal budget paces rereads, so a member the knot has never read mustn't wait \
421 its turn behind it and be refused at the handshake for the passes in between"
422 );
423
424 let spent = KeyTtl::from_secs(1).lease_from(UnixSeconds::new(0));
425 members
426 .iter()
427 .for_each(|member| _ = index.keys().record(member, Vec::new(), spent));
428 seen.lock().unwrap().clear();
429
430 futures::stream::iter(0..3)
431 .for_each(|_| async {
432 fill_once(&index, &atproto, &Slots::testing(4), &pacer, pace, &cursor).await;
433 })
434 .await;
435 let order = seen.lock().unwrap().clone();
436 assert_eq!(
437 order.iter().map(String::as_str).collect::<Vec<_>>(),
438 vec![BAILEY, OLAREN, TEQ],
439 "one member per pass in turn, or members whose PDS stays down keep the front of \
440 the queue and the knot never reads the rest"
441 );
442}