+198 -57

knot2/crates/knot-atproto/src/lib.rs

View file Reviewed

··· 54 54 use http::StatusCode; 55 55 use knot_cache::{ 56 56 Admitted, AsyncCache, EntryCount, Expiring, GroupQuota, MokaFuture, Quotas, Rejected, 57 57 - TotalQuota, 57 57 + TotalQuota, Weight, 58 58 }; 59 59 use knot_runtime::{ 60 60 Clock, DnsTxtResolver, HttpRequest, HttpTransport, NetworkError, PublicKeyBytes, SystemDns, ··· 76 76 Nsid::new_static("sh.tangled.repo").expect("literal nsid parses") 77 77 } 78 78 const PUBKEY_MAX_PAGES: usize = 8; 79 79 + const PUBKEY_TTL: Duration = Duration::from_secs(30); 80 80 + const MAX_PUBKEY_CACHE_BYTES: u64 = 1 << 20; 81 81 + const PUBKEY_CACHE_ENTRY_OVERHEAD: u64 = 128; 82 82 + const PUBKEY_CACHE_KEY_OVERHEAD: u64 = 192; 79 83 const MAX_IDENTITY_CACHE: usize = 4096; 80 84 const MAX_SEEN_JTI: usize = 8192; 81 85 ··· 135 139 _ => false, 136 140 } 137 141 } 142 142 + 143 143 + pub fn is_gone(&self) -> bool { 144 144 + match self { 145 145 + AtprotoError::Resolve(error) => error.is_gone(), 146 146 + AtprotoError::Jwt(_) 147 147 + | AtprotoError::Network(_) 148 148 + | AtprotoError::ListRecords { .. } 149 149 + | AtprotoError::MalformedRecords(_) 150 150 + | AtprotoError::BadPdsEndpoint { .. } 151 151 + | AtprotoError::Replay { .. } 152 152 + | AtprotoError::ReplayStoreSaturated 153 153 + | AtprotoError::ReplayShareExhausted { .. } 154 154 + | AtprotoError::Identity(_) 155 155 + | AtprotoError::PlcSubmit { .. } 156 156 + | AtprotoError::PutRecord { .. } 157 157 + | AtprotoError::GetRecord { .. } 158 158 + | AtprotoError::PointerEncode(_) 159 159 + | AtprotoError::MalformedReceipt(_) => false, 160 160 + } 161 161 + } 138 162 } 139 163 140 164 #[derive(Clone)] ··· 157 181 expires_at: UnixMicros, 158 182 } 159 183 184 184 + #[derive(Clone)] 185 185 + pub enum ClaimedKeys { 186 186 + Published(Vec<OfferedKey>), 187 187 + Unread(Arc<AtprotoError>), 188 188 + } 189 189 + 190 190 + fn claimed_weight(cached: &Cached<ClaimedKeys>) -> Weight { 191 191 + Weight::new(match &cached.resolution { 192 192 + ClaimedKeys::Published(keys) => keys 193 193 + .iter() 194 194 + .map(|key| key.as_bytes().len() as u64 + PUBKEY_CACHE_KEY_OVERHEAD) 195 195 + .sum::<u64>() 196 196 + .saturating_add(PUBKEY_CACHE_ENTRY_OVERHEAD), 197 197 + ClaimedKeys::Unread(_) => PUBKEY_CACHE_ENTRY_OVERHEAD, 198 198 + }) 199 199 + } 200 200 + 160 201 pub struct Atproto<H, C> { 161 202 http: H, 162 203 clock: C, ··· 165 206 dns: Arc<dyn DnsTxtResolver>, 166 207 identities: MokaFuture<AccountDid, Cached<Resolution>>, 167 208 handles: MokaFuture<Handle, Cached<HandleResolution>>, 209 209 + claimed: MokaFuture<AccountDid, Cached<ClaimedKeys>>, 168 210 seen_jti: Expiring<(AccountDid, JwtNonce), AccountDid, ()>, 169 211 } 170 212 ··· 178 220 dns: Arc::new(SystemDns::new()), 179 221 identities: MokaFuture::by_count(EntryCount::new(MAX_IDENTITY_CACHE as u64)), 180 222 handles: MokaFuture::by_count(EntryCount::new(MAX_IDENTITY_CACHE as u64)), 223 223 + claimed: MokaFuture::by_weight(Weight::new(MAX_PUBKEY_CACHE_BYTES), claimed_weight), 181 224 seen_jti: Expiring::new(Quotas { 182 225 per_group: GroupQuota::new(MAX_JTI_PER_ISSUER), 183 226 total: TotalQuota::new(MAX_SEEN_JTI), ··· 212 255 match filled.value.resolution { 213 256 Resolution::Found(identity) => Ok(identity), 214 257 Resolution::Transient(error) => Err(error), 215 215 - Resolution::Failed(error) if fresh => Err(error), 258 258 + Resolution::Failed(error) if fresh || error.is_gone() => Err(error), 216 259 Resolution::Failed(_) => Err(ResolveError::RecentlyFailed { did: did.clone() }), 217 260 } 218 261 } ··· 385 428 .await 386 429 .map_err(ResolveError::from)?; 387 430 if !response.status.is_success() { 388 388 - return Err(ResolveError::Status { 389 389 - status: HttpStatus::from(response.status), 390 390 - }); 431 431 + let status = HttpStatus::from(response.status); 432 432 + return match status.get() { 433 433 + 404 | 410 => Err(ResolveError::Gone { 434 434 + did: did.clone(), 435 435 + status, 436 436 + }), 437 437 + _ => Err(ResolveError::Status { status }), 438 438 + }; 391 439 } 392 440 resolve::identity_from_document(did, &response.body) 393 441 } 394 442 443 443 + pub fn document_url(&self, did: &AccountDid) -> Result<Url, ResolveError> { 444 444 + resolve::document_url(did, &self.plc_directory) 445 445 + } 446 446 + 395 447 pub async fn resolve_pubkeys(&self, did: &AccountDid) -> Result<Vec<OfferedKey>, AtprotoError> { 396 448 let identity = self.resolve_identity(did).await?; 449 449 + self.pubkeys_at(&identity, did).await 450 450 + } 451 451 + 452 452 + pub async fn claimed_pubkeys(&self, did: &AccountDid) -> ClaimedKeys { 453 453 + let now = self.clock.now_unix_micros(); 454 454 + let filled = self 455 455 + .claimed 456 456 + .get_or_fill_if( 457 457 + did.clone(), 458 458 + |cached: &Cached<ClaimedKeys>| cached.expires_at <= now, 459 459 + self.fill_claimed(did, now), 460 460 + ) 461 461 + .await; 462 462 + filled.value.resolution 463 463 + } 464 464 + 465 465 + async fn fill_claimed(&self, did: &AccountDid, now: UnixMicros) -> Cached<ClaimedKeys> { 466 466 + match self.resolve_pubkeys(did).await { 467 467 + Ok(keys) => Cached { 468 468 + resolution: ClaimedKeys::Published(keys), 469 469 + expires_at: expires(now, PUBKEY_TTL), 470 470 + }, 471 471 + Err(error) => Cached { 472 472 + expires_at: match error.is_transient() { 473 473 + true => now, 474 474 + false => expires(now, NEGATIVE_TTL), 475 475 + }, 476 476 + resolution: ClaimedKeys::Unread(Arc::new(error)), 477 477 + }, 478 478 + } 479 479 + } 480 480 + 481 481 + pub async fn pubkeys_at( 482 482 + &self, 483 483 + identity: &Identity, 484 484 + did: &AccountDid, 485 485 + ) -> Result<Vec<OfferedKey>, AtprotoError> { 397 486 let http = &self.http; 398 487 let pds = &identity.pds; 399 488 let pages = stream::try_unfold(Page::First(PUBKEY_MAX_PAGES), move |state| async move { ··· 600 689 ResolveError::Status { status } => { 601 690 (400..500).contains(&status.get()) && status.get() != 429 602 691 } 603 603 - ResolveError::Malformed(_) 692 692 + ResolveError::Gone { .. } 693 693 + | ResolveError::Malformed(_) 604 694 | ResolveError::IdMismatch { .. } 605 695 | ResolveError::BadSigningKey(_) 606 696 | ResolveError::BadPds { .. } => true, ··· 686 776 use futures::StreamExt; 687 777 use http::StatusCode; 688 778 use knot_runtime::{DnsTxtResolver, FakeDns, FakeHttp, ManualClock, NetworkError}; 689 689 - use std::sync::Arc; 690 779 use std::sync::atomic::{AtomicUsize, Ordering}; 780 780 + use std::sync::{Arc, Mutex}; 691 781 692 782 const POINTER_RKEY: &str = "3jzfcijpj2z2a"; 693 783 const POINTER_CID: &str = "bafyreidfayvfuwqa7qlnopdjiqrxzs6blmoeu4rujcjtnci5beludirz2a"; ··· 962 1052 let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc()); 963 1053 let first = atproto.resolve_identity(&did(SQUID)).await.unwrap_err(); 964 1054 assert!( 965 965 - matches!(first, AtprotoError::Resolve(ResolveError::Status { status }) if status.get() == 404), 1055 1055 + matches!(first, AtprotoError::Resolve(ResolveError::Gone { status, .. }) if status.get() == 404), 966 1056 "got {first:?}" 967 1057 ); 968 1058 let second = atproto.resolve_identity(&did(SQUID)).await.unwrap_err(); 969 1059 assert!( 970 970 - matches!( 971 971 - second, 972 972 - AtprotoError::Resolve(ResolveError::RecentlyFailed { .. }) 973 973 - ), 974 974 - "got {second:?}" 1060 1060 + matches!(second, AtprotoError::Resolve(ResolveError::Gone { .. })), 1061 1061 + "a caller that acts on a missing account must see the same answer from the cache as \ 1062 1062 + from the fetch, or it acts on the first read only: got {second:?}" 975 1063 ); 976 1064 assert_eq!( 977 1065 hits.load(Ordering::SeqCst), ··· 1117 1205 } 1118 1206 1119 1207 #[tokio::test] 1120 1120 - async fn pubkeys_are_fetched_from_the_resolved_pds() { 1121 1121 - let signing = signer(9); 1122 1122 - let line = ssh_line("ssh-ed25519", &[4u8; 32], "nel@oyster.cafe"); 1123 1123 - let expected = parse_authorized_key(&line).unwrap(); 1124 1124 - let body = list_body(&[line], None); 1125 1125 - let http = FakeHttp::new(move |request| { 1126 1126 - if request.url.path().ends_with("did.json") 1127 1127 - || request.url.host_str() == Some("plc.directory") 1128 1128 - { 1129 1129 - Ok(ok(squid_doc(&signing))) 1130 1130 - } else { 1131 1131 - assert_eq!(request.url.host_str(), Some("pds.oyster.cafe")); 1132 1132 - assert!(request.url.path().ends_with("com.atproto.repo.listRecords")); 1133 1133 - assert!( 1134 1134 - request 1135 1135 - .url 1136 1136 - .query() 1137 1137 - .unwrap() 1138 1138 - .contains("sh.tangled.publicKey") 1139 1139 - ); 1140 1140 - Ok(ok(body.clone())) 1141 1141 - } 1142 1142 - }); 1143 1143 - let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc()); 1144 1144 - let keys = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap(); 1145 1145 - assert_eq!(keys, vec![expected]); 1146 1146 - } 1147 1147 - 1148 1148 - #[tokio::test] 1149 1208 async fn pubkey_resolution_follows_the_cursor() { 1150 1209 let doc_key = signer(9); 1151 1210 let list_hits = Arc::new(AtomicUsize::new(0)); ··· 1171 1230 let keys = atproto.resolve_pubkeys(&did(SQUID)).await.unwrap(); 1172 1231 assert_eq!(keys.len(), 2); 1173 1232 assert_eq!(list_hits.load(Ordering::SeqCst), 2); 1233 1233 + } 1234 1234 + 1235 1235 + #[tokio::test] 1236 1236 + async fn a_repeated_claim_is_served_from_the_cache_instead_of_the_pds() { 1237 1237 + let signing = signer(9); 1238 1238 + let line = ssh_line("ssh-ed25519", &[4u8; 32], "nel@oyster.cafe"); 1239 1239 + let expected = parse_authorized_key(&line).unwrap(); 1240 1240 + let listings = Arc::new(AtomicUsize::new(0)); 1241 1241 + let counter = listings.clone(); 1242 1242 + let body = list_body(&[line], None); 1243 1243 + let http = FakeHttp::new(move |request| { 1244 1244 + if request.url.host_str() == Some("plc.directory") { 1245 1245 + return Ok(ok(squid_doc(&signing))); 1246 1246 + } 1247 1247 + assert_eq!(request.url.host_str(), Some("pds.oyster.cafe")); 1248 1248 + assert!(request.url.path().ends_with("com.atproto.repo.listRecords")); 1249 1249 + assert!( 1250 1250 + request 1251 1251 + .url 1252 1252 + .query() 1253 1253 + .unwrap() 1254 1254 + .contains("sh.tangled.publicKey") 1255 1255 + ); 1256 1256 + counter.fetch_add(1, Ordering::SeqCst); 1257 1257 + Ok(ok(body.clone())) 1258 1258 + }); 1259 1259 + let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc()); 1260 1260 + 1261 1261 + let first = atproto.claimed_pubkeys(&did(SQUID)).await; 1262 1262 + assert!( 1263 1263 + matches!(&first, ClaimedKeys::Published(keys) if keys == std::slice::from_ref(&expected)) 1264 1264 + ); 1265 1265 + let second = atproto.claimed_pubkeys(&did(SQUID)).await; 1266 1266 + assert!(matches!(&second, ClaimedKeys::Published(keys) if keys == &[expected])); 1267 1267 + assert_eq!( 1268 1268 + listings.load(Ordering::SeqCst), 1269 1269 + 1, 1270 1270 + "an ssh handshake asserting a login name reads the claimed account's records, so a \ 1271 1271 + repeat inside the cache turn mustn't read them again, or any stranger with an ssh \ 1272 1272 + client can make the knot fetch from that account's PDS at will" 1273 1273 + ); 1274 1274 + 1275 1275 + atproto.clock.advance(PUBKEY_TTL + Duration::from_micros(1)); 1276 1276 + let _ = atproto.claimed_pubkeys(&did(SQUID)).await; 1277 1277 + assert_eq!( 1278 1278 + listings.load(Ordering::SeqCst), 1279 1279 + 2, 1280 1280 + "a key published after the last read will be read once the cache turn is over" 1281 1281 + ); 1282 1282 + } 1283 1283 + 1284 1284 + #[tokio::test] 1285 1285 + async fn a_refused_claim_read_is_cached_and_an_outage_is_retried() { 1286 1286 + let refusal = Arc::new(Mutex::new(StatusCode::BAD_REQUEST)); 1287 1287 + let answered = Arc::clone(&refusal); 1288 1288 + let signing = signer(9); 1289 1289 + let listings = Arc::new(AtomicUsize::new(0)); 1290 1290 + let counter = listings.clone(); 1291 1291 + let http = FakeHttp::new(move |request| { 1292 1292 + if request.url.host_str() == Some("plc.directory") { 1293 1293 + return Ok(ok(squid_doc(&signing))); 1294 1294 + } 1295 1295 + counter.fetch_add(1, Ordering::SeqCst); 1296 1296 + Ok(status(*answered.lock().unwrap(), Bytes::new())) 1297 1297 + }); 1298 1298 + let atproto = Atproto::new(http, clock(), knot_did(KNOT), plc()); 1299 1299 + 1300 1300 + assert!(matches!( 1301 1301 + atproto.claimed_pubkeys(&did(SQUID)).await, 1302 1302 + ClaimedKeys::Unread(_) 1303 1303 + )); 1304 1304 + let _ = atproto.claimed_pubkeys(&did(SQUID)).await; 1305 1305 + assert_eq!( 1306 1306 + listings.load(Ordering::SeqCst), 1307 1307 + 1, 1308 1308 + "a listing the PDS refuses outright is served from the negative cache, so \ 1309 1309 + repeating the claim mustn't read the PDS again" 1310 1310 + ); 1311 1311 + 1312 1312 + atproto 1313 1313 + .clock 1314 1314 + .advance(NEGATIVE_TTL + Duration::from_micros(1)); 1315 1315 + *refusal.lock().unwrap() = StatusCode::SERVICE_UNAVAILABLE; 1316 1316 + let _ = atproto.claimed_pubkeys(&did(SQUID)).await; 1317 1317 + let _ = atproto.claimed_pubkeys(&did(SQUID)).await; 1318 1318 + assert_eq!( 1319 1319 + listings.load(Ordering::SeqCst), 1320 1320 + 3, 1321 1321 + "a transient failure is cached with no lifetime, so the account's next claim will \ 1322 1322 + read the PDS again instead of waiting out a negative turn" 1323 1323 + ); 1174 1324 } 1175 1325 1176 1326 #[tokio::test] ··· 1808 1958 } 1809 1959 1810 1960 #[tokio::test] 1811 1811 - async fn a_coalesced_404_wave_gives_one_caller_the_real_error_and_masks_the_rest() { 1961 1961 + async fn every_caller_in_a_coalesced_404_wave_learns_the_account_is_missing() { 1812 1962 let (results, calls) = gated_wave(StatusCode::NOT_FOUND, Bytes::new()).await; 1813 1963 assert_eq!(calls, 1, "404 wave coalesces into one outbound fetch"); 1814 1814 - let real = results 1964 1964 + let gone = results 1815 1965 .iter() 1816 1966 .filter(|outcome| { 1817 1967 matches!( 1818 1968 outcome, 1819 1819 - Err(AtprotoError::Resolve(ResolveError::Status { status })) if status.get() == 404 1969 1969 + Err(AtprotoError::Resolve(ResolveError::Gone { status, .. })) if status.get() == 404 1820 1970 ) 1821 1971 }) 1822 1972 .count(); 1823 1823 - let masked = results 1824 1824 - .iter() 1825 1825 - .filter(|outcome| { 1826 1826 - matches!( 1827 1827 - outcome, 1828 1828 - Err(AtprotoError::Resolve(ResolveError::RecentlyFailed { .. })) 1829 1829 - ) 1830 1830 - }) 1831 1831 - .count(); 1832 1832 - assert_eq!(real, 1, "exactly one caller observes the real 404"); 1833 1973 assert_eq!( 1834 1834 - masked, 1999, 1835 1835 - "the rest of the coalesced wave is masked as RecentlyFailed" 1974 1974 + gone, 2000, 1975 1975 + "coalescing mustn't decide which callers learn the account is missing, since the \ 1976 1976 + callers served from the cache act on the answer the same way" 1836 1977 ); 1837 1978 } 1838 1979

+66

knot2/crates/knot-atproto/src/resolve.rs

View file Reviewed

··· 39 39 BadPlcDirectory { value: String }, 40 40 #[error("identity {did} recently failed to resolve and is negatively cached")] 41 41 RecentlyFailed { did: AccountDid }, 42 42 + #[error("{did} doesn't have a DID document, the directory answered HTTP {status}")] 43 43 + Gone { did: AccountDid, status: HttpStatus }, 42 44 #[error("did:web document for {did} doesn't publish expected signing key")] 43 45 ExpectedKeyAbsent { did: RepoDid }, 44 46 #[error("handle {handle} has no atproto DNS or well-known record")] ··· 63 65 ResolveError::Network(_) => true, 64 66 ResolveError::Status { status } => status.is_transient(), 65 67 _ => false, 68 68 + } 69 69 + } 70 70 + 71 71 + pub fn is_gone(&self) -> bool { 72 72 + match self { 73 73 + ResolveError::Gone { .. } | ResolveError::Unresolvable { .. } => true, 74 74 + ResolveError::UnsupportedMethod { .. } 75 75 + | ResolveError::Status { .. } 76 76 + | ResolveError::Network(_) 77 77 + | ResolveError::Malformed(_) 78 78 + | ResolveError::IdMismatch { .. } 79 79 + | ResolveError::InsecureScheme { .. } 80 80 + | ResolveError::BlockedHost { .. } 81 81 + | ResolveError::MissingSigningKey 82 82 + | ResolveError::BadSigningKey(_) 83 83 + | ResolveError::MissingPds 84 84 + | ResolveError::BadPds { .. } 85 85 + | ResolveError::BadPlcDirectory { .. } 86 86 + | ResolveError::RecentlyFailed { .. } 87 87 + | ResolveError::ExpectedKeyAbsent { .. } 88 88 + | ResolveError::HandleUnresolvable { .. } 89 89 + | ResolveError::HandleAmbiguous { .. } 90 90 + | ResolveError::HandleForwardMalformed { .. } 91 91 + | ResolveError::HandleMismatch { .. } 92 92 + | ResolveError::HandleRecentlyFailed { .. } => false, 66 93 } 67 94 } 68 95 } ··· 582 609 case.name 583 610 ); 584 611 }); 612 612 + } 613 613 + 614 614 + #[test] 615 615 + fn only_a_missing_document_reads_as_an_account_that_no_longer_exists() { 616 616 + assert!( 617 617 + ResolveError::Gone { 618 618 + did: did("did:plc:squid"), 619 619 + status: HttpStatus::new(404) 620 620 + } 621 621 + .is_gone() 622 622 + ); 623 623 + assert!( 624 624 + ResolveError::Unresolvable { 625 625 + value: "did:web:nel.pet/../..".to_string() 626 626 + } 627 627 + .is_gone(), 628 628 + "a DID that doesn't form a document location can never have published a key" 629 629 + ); 630 630 + assert!( 631 631 + !ResolveError::Malformed("{".to_string()).is_gone(), 632 632 + "a knot that read a broken document as proof the account stopped publishing keys \ 633 633 + would delete the keys of everyone behind one bad PDS migration" 634 634 + ); 635 635 + assert!( 636 636 + !ResolveError::RecentlyFailed { 637 637 + did: did("did:plc:squid") 638 638 + } 639 639 + .is_gone(), 640 640 + "the negative cache stores broken documents as well as missing documents, so \ 641 641 + reading it back mustn't stand in for either" 642 642 + ); 643 643 + assert!( 644 644 + !ResolveError::Status { 645 645 + status: HttpStatus::new(404) 646 646 + } 647 647 + .is_gone(), 648 648 + "the handle and pubkey fetches answer 404 too, so the knot can't read a bare 404 \ 649 649 + as an account that stopped existing" 650 650 + ); 585 651 } 586 652 }

+12

knot2/crates/knot-cache/src/lib.rs

View file Reviewed

··· 365 365 } 366 366 } 367 367 368 368 + pub fn by_weight<F>(max_weight: Weight, weigh: F) -> Self 369 369 + where 370 370 + F: Fn(&V) -> Weight + Send + Sync + 'static, 371 371 + { 372 372 + Self { 373 373 + inner: moka::future::Cache::builder() 374 374 + .max_capacity(max_weight.get()) 375 375 + .weigher(move |_key: &K, value: &V| weigh(value).get().min(u32::MAX as u64) as u32) 376 376 + .build_with_hasher(DeterministicHasher::default()), 377 377 + } 378 378 + } 379 379 + 368 380 pub async fn get_or_fill_if<Fut, P>(&self, key: K, refill_if: P, fill: Fut) -> Filled<V> 369 381 where 370 382 Fut: Future<Output = V> + Send,