colibri.social / appview
[READ-ONLY] Mirror of https://github.com/colibri-social/appview. The AppView (backend server) for Colibri api.colibri.social
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

feat: better image loading (#69)

author
Louis Escher committer
GitHub date (Jul 25, 2026, 5:23 PM +0200) commit 4424fea8 parent 647c72a4
+856 -54
+163
Cargo.lock
··· 975 975 checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" 976 976 977 977 [[package]] 978 + name = "byteorder-lite" 979 + version = "0.1.0" 980 + source = "registry+https://github.com/rust-lang/crates.io-index" 981 + checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" 982 + 983 + [[package]] 978 984 name = "bytes" 979 985 version = "1.11.1" 980 986 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1152 1158 "futures", 1153 1159 "futures-util", 1154 1160 "hex", 1161 + "image", 1155 1162 "jsonwebtoken", 1156 1163 "k256", 1157 1164 "log", ··· 1181 1188 ] 1182 1189 1183 1190 [[package]] 1191 + name = "color_quant" 1192 + version = "1.1.0" 1193 + source = "registry+https://github.com/rust-lang/crates.io-index" 1194 + checksum = "3d7b894f5411737b7867f4827955924d7c254fc9f4d91a6aad6b097804b1018b" 1195 + 1196 + [[package]] 1184 1197 name = "colorchoice" 1185 1198 version = "1.0.5" 1186 1199 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1339 1352 version = "2.4.0" 1340 1353 source = "registry+https://github.com/rust-lang/crates.io-index" 1341 1354 checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" 1355 + 1356 + [[package]] 1357 + name = "crc32fast" 1358 + version = "1.5.0" 1359 + source = "registry+https://github.com/rust-lang/crates.io-index" 1360 + checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" 1361 + dependencies = [ 1362 + "cfg-if", 1363 + ] 1342 1364 1343 1365 [[package]] 1344 1366 name = "crossbeam-queue" ··· 1926 1948 checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" 1927 1949 1928 1950 [[package]] 1951 + name = "fax" 1952 + version = "0.2.7" 1953 + source = "registry+https://github.com/rust-lang/crates.io-index" 1954 + checksum = "caf1079563223d5d59d83c85886a56e586cfd5c1a26292e971a0fa266531ac5a" 1955 + 1956 + [[package]] 1957 + name = "fdeflate" 1958 + version = "0.3.7" 1959 + source = "registry+https://github.com/rust-lang/crates.io-index" 1960 + checksum = "1e6853b52649d4ac5c0bd02320cddc5ba956bdb407c4b75a2c6b75bf51500f8c" 1961 + dependencies = [ 1962 + "simd-adler32", 1963 + ] 1964 + 1965 + [[package]] 1929 1966 name = "ff" 1930 1967 version = "0.13.1" 1931 1968 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 1974 2011 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80" 1975 2012 1976 2013 [[package]] 2014 + name = "flate2" 2015 + version = "1.1.9" 2016 + source = "registry+https://github.com/rust-lang/crates.io-index" 2017 + checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" 2018 + dependencies = [ 2019 + "crc32fast", 2020 + "miniz_oxide", 2021 + ] 2022 + 2023 + [[package]] 1977 2024 name = "flume" 1978 2025 version = "0.11.1" 1979 2026 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 2282 2329 dependencies = [ 2283 2330 "opaque-debug", 2284 2331 "polyval", 2332 + ] 2333 + 2334 + [[package]] 2335 + name = "gif" 2336 + version = "0.14.2" 2337 + source = "registry+https://github.com/rust-lang/crates.io-index" 2338 + checksum = "ee8cfcc411d9adbbaba82fb72661cc1bcca13e8bba98b364e62b2dba8f960159" 2339 + dependencies = [ 2340 + "color_quant", 2341 + "weezl", 2285 2342 ] 2286 2343 2287 2344 [[package]] ··· 2878 2935 ] 2879 2936 2880 2937 [[package]] 2938 + name = "image" 2939 + version = "0.25.10" 2940 + source = "registry+https://github.com/rust-lang/crates.io-index" 2941 + checksum = "85ab80394333c02fe689eaf900ab500fbd0c2213da414687ebf995a65d5a6104" 2942 + dependencies = [ 2943 + "bytemuck", 2944 + "byteorder-lite", 2945 + "color_quant", 2946 + "gif", 2947 + "image-webp", 2948 + "moxcms", 2949 + "num-traits", 2950 + "png", 2951 + "tiff", 2952 + "zune-core", 2953 + "zune-jpeg", 2954 + ] 2955 + 2956 + [[package]] 2957 + name = "image-webp" 2958 + version = "0.2.4" 2959 + source = "registry+https://github.com/rust-lang/crates.io-index" 2960 + checksum = "525e9ff3e1a4be2fbea1fdf0e98686a6d98b4d8f937e1bf7402245af1909e8c3" 2961 + dependencies = [ 2962 + "byteorder-lite", 2963 + "quick-error", 2964 + ] 2965 + 2966 + [[package]] 2881 2967 name = "impl-more" 2882 2968 version = "0.1.9" 2883 2969 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 3577 3663 checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" 3578 3664 dependencies = [ 3579 3665 "adler2", 3666 + "simd-adler32", 3580 3667 ] 3581 3668 3582 3669 [[package]] ··· 3616 3703 "ctutils", 3617 3704 "hybrid-array", 3618 3705 "num-traits", 3706 + ] 3707 + 3708 + [[package]] 3709 + name = "moxcms" 3710 + version = "0.8.1" 3711 + source = "registry+https://github.com/rust-lang/crates.io-index" 3712 + checksum = "bb85c154ba489f01b25c0d36ae69a87e4a1c73a72631fc6c0eb6dde34a73e44b" 3713 + dependencies = [ 3714 + "num-traits", 3715 + "pxfm", 3619 3716 ] 3620 3717 3621 3718 [[package]] ··· 4494 4591 ] 4495 4592 4496 4593 [[package]] 4594 + name = "png" 4595 + version = "0.18.1" 4596 + source = "registry+https://github.com/rust-lang/crates.io-index" 4597 + checksum = "60769b8b31b2a9f263dae2776c37b1b28ae246943cf719eb6946a1db05128a61" 4598 + dependencies = [ 4599 + "bitflags 2.11.0", 4600 + "crc32fast", 4601 + "fdeflate", 4602 + "flate2", 4603 + "miniz_oxide", 4604 + ] 4605 + 4606 + [[package]] 4497 4607 name = "polyval" 4498 4608 version = "0.6.2" 4499 4609 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 4642 4752 "quote", 4643 4753 "syn 1.0.109", 4644 4754 ] 4755 + 4756 + [[package]] 4757 + name = "pxfm" 4758 + version = "0.1.30" 4759 + source = "registry+https://github.com/rust-lang/crates.io-index" 4760 + checksum = "d55d956fa96f5ec02be2e13af0e20391a5aa83d6a074e3ad368959d0fab299ea" 4761 + 4762 + [[package]] 4763 + name = "quick-error" 4764 + version = "2.0.1" 4765 + source = "registry+https://github.com/rust-lang/crates.io-index" 4766 + checksum = "a993555f31e5a609f617c12db6250dedcac1b0a85076912c436e6fc9b2c8e6a3" 4645 4767 4646 4768 [[package]] 4647 4769 name = "quinn" ··· 5918 6040 ] 5919 6041 5920 6042 [[package]] 6043 + name = "simd-adler32" 6044 + version = "0.3.10" 6045 + source = "registry+https://github.com/rust-lang/crates.io-index" 6046 + checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" 6047 + 6048 + [[package]] 5921 6049 name = "simdutf8" 5922 6050 version = "0.1.5" 5923 6051 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 6499 6627 checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" 6500 6628 dependencies = [ 6501 6629 "cfg-if", 6630 + ] 6631 + 6632 + [[package]] 6633 + name = "tiff" 6634 + version = "0.11.3" 6635 + source = "registry+https://github.com/rust-lang/crates.io-index" 6636 + checksum = "b63feaf3343d35b6ca4d50483f94843803b0f51634937cc2ec519fc32232bc52" 6637 + dependencies = [ 6638 + "fax", 6639 + "flate2", 6640 + "half", 6641 + "quick-error", 6642 + "weezl", 6643 + "zune-jpeg", 6502 6644 ] 6503 6645 6504 6646 [[package]] ··· 7383 7525 ] 7384 7526 7385 7527 [[package]] 7528 + name = "weezl" 7529 + version = "0.1.12" 7530 + source = "registry+https://github.com/rust-lang/crates.io-index" 7531 + checksum = "a28ac98ddc8b9274cb41bb4d9d4d5c425b6020c50c46f25559911905610b4a88" 7532 + 7533 + [[package]] 7386 7534 name = "whoami" 7387 7535 version = "1.6.1" 7388 7536 source = "registry+https://github.com/rust-lang/crates.io-index" ··· 7983 8131 version = "1.0.21" 7984 8132 source = "registry+https://github.com/rust-lang/crates.io-index" 7985 8133 checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" 8134 + 8135 + [[package]] 8136 + name = "zune-core" 8137 + version = "0.5.1" 8138 + source = "registry+https://github.com/rust-lang/crates.io-index" 8139 + checksum = "cb8a0807f7c01457d0379ba880ba6322660448ddebc890ce29bb64da71fb40f9" 8140 + 8141 + [[package]] 8142 + name = "zune-jpeg" 8143 + version = "0.5.15" 8144 + source = "registry+https://github.com/rust-lang/crates.io-index" 8145 + checksum = "27bc9d5b815bc103f142aa054f561d9187d191692ec7c2d1e2b4737f8dbd7296" 8146 + dependencies = [ 8147 + "zune-core", 8148 + ]
+1
Cargo.toml
··· 46 46 web-push = { version = "0.11", default-features = false } 47 47 bytes = "1" 48 48 lru = "0.12" 49 + image = { version = "0.25.10", default-features = false, features = ["jpeg", "png", "gif", "webp", "bmp", "ico", "tiff"] } 49 50 50 51 [dev-dependencies] 51 52 wiremock = "0.6"
+12 -9
src/lib/blob_cache.rs
··· 1 1 //! A bounded in-memory LRU cache for blob bytes. 2 2 //! 3 3 //! Blobs are content-addressed by CID (immutable), so a fetched blob is valid 4 - //! forever and can be shared across DIDs. The PDS upstream doesn't honour HTTP 4 + //! forever and can be shared across DIDs. Entries are keyed by an opaque string 5 + //! built from that CID plus the rendition being served (see 6 + //! `get_blob_handler::cache_key`), so a resized avatar variant and the original 7 + //! bytes coexist without colliding. The PDS upstream doesn't honour HTTP 5 8 //! Range, so to serve cheap byte-ranges (seeking, media duration) the AppView 6 9 //! holds the full bytes locally and slices them. Keeping recently-served blobs 7 - //! in RAM means the PDS is hit at most once per CID while it stays hot, and 10 + //! in RAM means the PDS is hit at most once per key while it stays hot, and 8 11 //! every range request after that is a zero-copy `Bytes::slice`. 9 12 //! 10 13 //! Eviction is by total byte size (least-recently-used first), not entry count. ··· 31 34 total: u64, 32 35 } 33 36 34 - /// Thread-safe, size-bounded LRU of blob bytes keyed by CID. 37 + /// Thread-safe, size-bounded LRU of blob bytes keyed by CID and rendition. 35 38 pub struct BlobCache { 36 39 inner: Mutex<Inner>, 37 40 cap: u64, ··· 62 65 Self::new(cap) 63 66 } 64 67 65 - /// Returns the cached blob for `cid`, marking it most-recently-used. 66 - pub fn get(&self, cid: &str) -> Option<CacheEntry> { 67 - self.inner.lock().unwrap().map.get(cid).cloned() 68 + /// Returns the cached blob for `key`, marking it most-recently-used. 69 + pub fn get(&self, key: &str) -> Option<CacheEntry> { 70 + self.inner.lock().unwrap().map.get(key).cloned() 68 71 } 69 72 70 - /// Inserts (or replaces) `cid`'s blob, then evicts least-recently-used 73 + /// Inserts (or replaces) `key`'s blob, then evicts least-recently-used 71 74 /// entries until the cache is back within its byte ceiling. A blob larger 72 75 /// than the whole ceiling is not cached at all, so it can't evict 73 76 /// everything else for a single use. 74 - pub fn insert(&self, cid: &str, entry: CacheEntry) { 77 + pub fn insert(&self, key: &str, entry: CacheEntry) { 75 78 let size = entry.bytes.len() as u64; 76 79 if size > self.cap { 77 80 return; ··· 80 83 let mut guard = self.inner.lock().unwrap(); 81 84 let inner = &mut *guard; 82 85 83 - if let Some(old) = inner.map.put(cid.to_string(), entry) { 86 + if let Some(old) = inner.map.put(key.to_string(), entry) { 84 87 inner.total = inner.total.saturating_sub(old.bytes.len() as u64); 85 88 } 86 89 inner.total += size;
+123 -6
src/lib/embed_fetch.rs
··· 6 6 7 7 use std::collections::HashMap; 8 8 use std::net::{IpAddr, SocketAddr}; 9 + use std::num::NonZeroUsize; 10 + use std::sync::{LazyLock, Mutex}; 9 11 use std::time::Duration; 10 12 13 + use lru::LruCache; 11 14 use reqwest::{Client, Response, Url, redirect::Policy}; 12 15 use scraper::{Html, Selector}; 13 16 use serde::Serialize; 14 17 15 18 /// How long an outbound fetch may take before it's abandoned. 16 19 const FETCH_TIMEOUT: Duration = Duration::from_secs(5); 20 + /// The budget for fetching a blob, which moves a real payload (an avatar, an 21 + /// attachment) rather than a page of HTML 22 + pub const BLOB_FETCH_TIMEOUT: Duration = Duration::from_secs(30); 17 23 /// Maximum number of redirects we'll follow (each one is re-validated). 18 24 const MAX_REDIRECTS: usize = 5; 25 + /// How many pinned clients to keep alive. Each one owns a connection pool, so 26 + /// this is really "how many upstream hosts stay warm at once". 27 + const CLIENT_CACHE_CAPACITY: usize = 256; 19 28 /// A browser-ish UA — many sites only emit OG tags for "real" user agents. 20 29 const USER_AGENT: &str = "Mozilla/5.0 (compatible; ColibriBot/1.0; +https://colibri.social)"; 21 30 ··· 165 174 Ok(addrs) 166 175 } 167 176 168 - fn pinned_client(host: &str, addrs: &[SocketAddr]) -> Result<Client, FetchError> { 169 - Client::builder() 170 - .timeout(FETCH_TIMEOUT) 177 + /// Identifies a pinned client by exactly the things that make it safe to reuse: 178 + /// the host it was built for, the addresses it is pinned to, and its timeout 179 + #[derive(Debug, Clone, PartialEq, Eq, Hash)] 180 + struct ClientKey { 181 + host: String, 182 + addrs: Vec<SocketAddr>, 183 + timeout: Duration, 184 + } 185 + 186 + static CLIENT_CACHE: LazyLock<Mutex<LruCache<ClientKey, Client>>> = LazyLock::new(|| { 187 + Mutex::new(LruCache::new( 188 + NonZeroUsize::new(CLIENT_CACHE_CAPACITY).expect("capacity is non-zero"), 189 + )) 190 + }); 191 + 192 + /// A client pinned to `addrs`, reused across calls. Building a `reqwest::Client` 193 + /// is not cheap, it constructs a fresh connection pool and loads the TLS root 194 + /// store, so doing it per request meant every blob fetch paid a full TLS 195 + /// handshake against a PDS we were already talking to. 196 + fn pinned_client( 197 + host: &str, 198 + addrs: &[SocketAddr], 199 + timeout: Duration, 200 + ) -> Result<Client, FetchError> { 201 + let mut pinned = addrs.to_vec(); 202 + pinned.sort(); 203 + 204 + let key = ClientKey { 205 + host: host.to_string(), 206 + addrs: pinned, 207 + timeout, 208 + }; 209 + 210 + if let Some(client) = CLIENT_CACHE.lock().unwrap().get(&key) { 211 + return Ok(client.clone()); 212 + } 213 + 214 + let client = Client::builder() 215 + .timeout(timeout) 171 216 .redirect(Policy::none()) 172 217 .user_agent(USER_AGENT) 218 + .pool_idle_timeout(Duration::from_secs(90)) 173 219 .resolve_to_addrs(host, addrs) 174 220 .build() 175 - .map_err(|e| FetchError::Upstream(e.to_string())) 221 + .map_err(|e| FetchError::Upstream(e.to_string()))?; 222 + 223 + CLIENT_CACHE.lock().unwrap().put(key, client.clone()); 224 + 225 + Ok(client) 176 226 } 177 227 178 228 /// SSRF-safe GET: validates the host (and every redirect target) against the 179 229 /// private-address blocklist, pins each hop's connection to the exact 180 230 /// addresses just validated, and follows up to [`MAX_REDIRECTS`] redirects 181 231 pub async fn guarded_get(raw_url: &str) -> Result<Response, FetchError> { 232 + guarded_get_with_timeout(raw_url, FETCH_TIMEOUT).await 233 + } 234 + 235 + /// [`guarded_get`] with an explicit time budget, for callers moving payloads 236 + /// that a scrape-sized timeout would cut off (see [`BLOB_FETCH_TIMEOUT`]). 237 + pub async fn guarded_get_with_timeout( 238 + raw_url: &str, 239 + timeout: Duration, 240 + ) -> Result<Response, FetchError> { 182 241 let mut current = Url::parse(raw_url).map_err(|e| FetchError::InvalidUrl(e.to_string()))?; 183 242 184 243 for _ in 0..=MAX_REDIRECTS { ··· 187 246 .host_str() 188 247 .ok_or_else(|| FetchError::InvalidUrl("missing host".into()))? 189 248 .to_string(); 190 - let client = pinned_client(&host, &addrs)?; 249 + let client = pinned_client(&host, &addrs, timeout)?; 191 250 192 251 let resp = client 193 252 .get(current.clone()) ··· 233 292 let host = url 234 293 .host_str() 235 294 .ok_or_else(|| FetchError::InvalidUrl("missing host".into()))?; 236 - let client = pinned_client(host, &addrs)?; 295 + let client = pinned_client(host, &addrs, FETCH_TIMEOUT)?; 237 296 Ok((client, url)) 238 297 } 239 298 ··· 368 427 mod tests { 369 428 use super::*; 370 429 use rocket::tokio; 430 + 431 + fn cached_clients_for(host: &str) -> usize { 432 + CLIENT_CACHE 433 + .lock() 434 + .unwrap() 435 + .iter() 436 + .filter(|(key, _)| key.host == host) 437 + .count() 438 + } 439 + 440 + #[test] 441 + fn repeated_fetches_to_one_host_share_a_client() { 442 + let host = "client-reuse.test"; 443 + let addrs = vec!["93.184.216.34:443".parse::<SocketAddr>().unwrap()]; 444 + 445 + for _ in 0..5 { 446 + pinned_client(host, &addrs, FETCH_TIMEOUT).unwrap(); 447 + } 448 + 449 + assert_eq!(cached_clients_for(host), 1); 450 + } 451 + 452 + /// The pin is the security boundary: a client validated against one set of 453 + /// addresses must never be handed to a request that resolved elsewhere. 454 + #[test] 455 + fn a_different_resolution_gets_a_different_client() { 456 + let host = "rebind.test"; 457 + let first = vec!["93.184.216.34:443".parse::<SocketAddr>().unwrap()]; 458 + let second = vec!["93.184.216.35:443".parse::<SocketAddr>().unwrap()]; 459 + 460 + pinned_client(host, &first, FETCH_TIMEOUT).unwrap(); 461 + pinned_client(host, &second, FETCH_TIMEOUT).unwrap(); 462 + 463 + assert_eq!(cached_clients_for(host), 2); 464 + } 465 + 466 + #[test] 467 + fn address_ordering_does_not_split_the_cache() { 468 + let host = "multi-a.test"; 469 + let a = "93.184.216.34:443".parse::<SocketAddr>().unwrap(); 470 + let b = "93.184.216.35:443".parse::<SocketAddr>().unwrap(); 471 + 472 + pinned_client(host, &[a, b], FETCH_TIMEOUT).unwrap(); 473 + pinned_client(host, &[b, a], FETCH_TIMEOUT).unwrap(); 474 + 475 + assert_eq!(cached_clients_for(host), 1); 476 + } 477 + 478 + #[test] 479 + fn timeouts_are_not_shared_between_clients() { 480 + let host = "timeouts.test"; 481 + let addrs = vec!["93.184.216.34:443".parse::<SocketAddr>().unwrap()]; 482 + 483 + pinned_client(host, &addrs, FETCH_TIMEOUT).unwrap(); 484 + pinned_client(host, &addrs, BLOB_FETCH_TIMEOUT).unwrap(); 485 + 486 + assert_eq!(cached_clients_for(host), 2); 487 + } 371 488 372 489 #[test] 373 490 fn blocks_loopback_and_private_ips() {
+283
src/lib/image_variant.rs
··· 1 + use std::io::Cursor; 2 + 3 + use bytes::Bytes; 4 + use image::{DynamicImage, ImageEncoder, ImageReader, imageops::FilterType}; 5 + 6 + pub const JPEG_QUALITY: u8 = 82; 7 + 8 + #[derive(Debug, Clone, Copy, PartialEq, Eq)] 9 + pub enum Variant { 10 + Small, 11 + Base, 12 + Large, 13 + } 14 + 15 + impl Variant { 16 + pub const ALL: [Variant; 3] = [Variant::Small, Variant::Base, Variant::Large]; 17 + 18 + pub fn parse(raw: &str) -> Option<Self> { 19 + match raw { 20 + "small" => Some(Variant::Small), 21 + "base" => Some(Variant::Base), 22 + "large" => Some(Variant::Large), 23 + _ => None, 24 + } 25 + } 26 + 27 + pub fn edge(self) -> u32 { 28 + match self { 29 + Variant::Small => 48, 30 + Variant::Base => 96, 31 + Variant::Large => 160, 32 + } 33 + } 34 + 35 + pub fn as_str(self) -> &'static str { 36 + match self { 37 + Variant::Small => "small", 38 + Variant::Base => "base", 39 + Variant::Large => "large", 40 + } 41 + } 42 + } 43 + 44 + pub struct Rendered { 45 + pub bytes: Bytes, 46 + pub content_type: String, 47 + } 48 + 49 + #[derive(Debug, thiserror::Error)] 50 + pub enum RenderError { 51 + #[error("unsupported image data: {0}")] 52 + Decode(String), 53 + #[error("failed to encode variant: {0}")] 54 + Encode(String), 55 + } 56 + 57 + pub fn is_resizable(content_type: &str) -> bool { 58 + let base = content_type 59 + .split(';') 60 + .next() 61 + .unwrap_or_default() 62 + .trim() 63 + .to_ascii_lowercase(); 64 + 65 + matches!( 66 + base.as_str(), 67 + "image/jpeg" 68 + | "image/jpg" 69 + | "image/png" 70 + | "image/webp" 71 + | "image/gif" 72 + | "image/bmp" 73 + | "image/x-icon" 74 + | "image/vnd.microsoft.icon" 75 + | "image/tiff" 76 + ) 77 + } 78 + 79 + /// Decodes `bytes` once and derives every [`Variant`] from it, so a single 80 + /// upstream fetch and a single decode serve an avatar at all the sizes the 81 + /// client asks for. 82 + pub fn render_all(bytes: &[u8]) -> Result<Vec<(Variant, Rendered)>, RenderError> { 83 + let decoded = decode(bytes)?; 84 + 85 + Variant::ALL 86 + .into_iter() 87 + .map(|variant| render_decoded(&decoded, variant).map(|rendered| (variant, rendered))) 88 + .collect() 89 + } 90 + 91 + fn decode(bytes: &[u8]) -> Result<DynamicImage, RenderError> { 92 + ImageReader::new(Cursor::new(bytes)) 93 + .with_guessed_format() 94 + .map_err(|e| RenderError::Decode(e.to_string()))? 95 + .decode() 96 + .map_err(|e| RenderError::Decode(e.to_string())) 97 + } 98 + 99 + fn render_decoded(decoded: &DynamicImage, variant: Variant) -> Result<Rendered, RenderError> { 100 + let edge = variant.edge(); 101 + let resized = if decoded.width() <= edge && decoded.height() <= edge { 102 + decoded.clone() 103 + } else { 104 + decoded.resize_to_fill(edge, edge, FilterType::Triangle) 105 + }; 106 + 107 + if has_alpha(&resized) { 108 + encode_png(&resized) 109 + } else { 110 + encode_jpeg(&resized) 111 + } 112 + } 113 + 114 + fn has_alpha(image: &DynamicImage) -> bool { 115 + image.color().has_alpha() 116 + } 117 + 118 + fn encode_png(image: &DynamicImage) -> Result<Rendered, RenderError> { 119 + let rgba = image.to_rgba8(); 120 + let mut out = Vec::new(); 121 + 122 + image::codecs::png::PngEncoder::new_with_quality( 123 + &mut out, 124 + image::codecs::png::CompressionType::Best, 125 + image::codecs::png::FilterType::Adaptive, 126 + ) 127 + .write_image( 128 + rgba.as_raw(), 129 + rgba.width(), 130 + rgba.height(), 131 + image::ExtendedColorType::Rgba8, 132 + ) 133 + .map_err(|e| RenderError::Encode(e.to_string()))?; 134 + 135 + Ok(Rendered { 136 + bytes: Bytes::from(out), 137 + content_type: String::from("image/png"), 138 + }) 139 + } 140 + 141 + fn encode_jpeg(image: &DynamicImage) -> Result<Rendered, RenderError> { 142 + let rgb = image.to_rgb8(); 143 + let mut out = Vec::new(); 144 + 145 + image::codecs::jpeg::JpegEncoder::new_with_quality(&mut out, JPEG_QUALITY) 146 + .write_image( 147 + rgb.as_raw(), 148 + rgb.width(), 149 + rgb.height(), 150 + image::ExtendedColorType::Rgb8, 151 + ) 152 + .map_err(|e| RenderError::Encode(e.to_string()))?; 153 + 154 + Ok(Rendered { 155 + bytes: Bytes::from(out), 156 + content_type: String::from("image/jpeg"), 157 + }) 158 + } 159 + 160 + #[cfg(test)] 161 + mod tests { 162 + use super::*; 163 + use image::{ImageFormat, Rgb, RgbImage, Rgba, RgbaImage}; 164 + 165 + fn render(bytes: &[u8], variant: Variant) -> Result<Rendered, RenderError> { 166 + render_decoded(&decode(bytes)?, variant) 167 + } 168 + 169 + fn opaque_source(size: u32) -> Vec<u8> { 170 + let mut img = RgbImage::new(size, size); 171 + for (x, y, px) in img.enumerate_pixels_mut() { 172 + *px = Rgb([(x % 256) as u8, (y % 256) as u8, 128]); 173 + } 174 + let mut out = Vec::new(); 175 + DynamicImage::ImageRgb8(img) 176 + .write_to(&mut Cursor::new(&mut out), ImageFormat::Png) 177 + .unwrap(); 178 + out 179 + } 180 + 181 + fn alpha_source(size: u32) -> Vec<u8> { 182 + let mut img = RgbaImage::new(size, size); 183 + for (x, y, px) in img.enumerate_pixels_mut() { 184 + *px = Rgba([(x % 256) as u8, (y % 256) as u8, 200, 128]); 185 + } 186 + let mut out = Vec::new(); 187 + DynamicImage::ImageRgba8(img) 188 + .write_to(&mut Cursor::new(&mut out), ImageFormat::Png) 189 + .unwrap(); 190 + out 191 + } 192 + 193 + fn dimensions(bytes: &[u8]) -> (u32, u32) { 194 + let img = ImageReader::new(Cursor::new(bytes)) 195 + .with_guessed_format() 196 + .unwrap() 197 + .decode() 198 + .unwrap(); 199 + (img.width(), img.height()) 200 + } 201 + 202 + #[test] 203 + fn parses_only_known_variants() { 204 + assert_eq!(Variant::parse("small"), Some(Variant::Small)); 205 + assert_eq!(Variant::parse("base"), Some(Variant::Base)); 206 + assert_eq!(Variant::parse("large"), Some(Variant::Large)); 207 + assert_eq!(Variant::parse("2048"), None); 208 + assert_eq!(Variant::parse(""), None); 209 + } 210 + 211 + #[test] 212 + fn downscales_to_the_variant_edge() { 213 + let rendered = render(&opaque_source(512), Variant::Base).unwrap(); 214 + assert_eq!(dimensions(&rendered.bytes), (96, 96)); 215 + } 216 + 217 + #[test] 218 + fn opaque_sources_encode_as_jpeg() { 219 + let rendered = render(&opaque_source(512), Variant::Small).unwrap(); 220 + assert_eq!(rendered.content_type, "image/jpeg"); 221 + } 222 + 223 + #[test] 224 + fn transparent_sources_keep_their_alpha_channel() { 225 + let rendered = render(&alpha_source(512), Variant::Small).unwrap(); 226 + assert_eq!(rendered.content_type, "image/png"); 227 + 228 + let decoded = ImageReader::new(Cursor::new(&rendered.bytes)) 229 + .with_guessed_format() 230 + .unwrap() 231 + .decode() 232 + .unwrap(); 233 + assert!(decoded.color().has_alpha()); 234 + } 235 + 236 + /// The point of the whole module: whatever the source weighs, a rendition 237 + /// is small in absolute terms. Real avatars run to hundreds of kilobytes. 238 + #[test] 239 + fn every_variant_of_a_large_source_stays_tiny() { 240 + let source = opaque_source(1024); 241 + 242 + for (variant, out) in render_all(&source).unwrap() { 243 + assert!( 244 + out.bytes.len() < 16 * 1024, 245 + "{} variant is {} bytes, source was {} bytes", 246 + variant.as_str(), 247 + out.bytes.len(), 248 + source.len() 249 + ); 250 + } 251 + } 252 + 253 + #[test] 254 + fn sources_smaller_than_the_edge_are_not_upscaled() { 255 + let rendered = render(&opaque_source(32), Variant::Large).unwrap(); 256 + assert_eq!(dimensions(&rendered.bytes), (32, 32)); 257 + } 258 + 259 + #[test] 260 + fn render_all_covers_every_variant_from_one_decode() { 261 + let rendered = render_all(&opaque_source(512)).unwrap(); 262 + 263 + assert_eq!(rendered.len(), Variant::ALL.len()); 264 + for (variant, out) in &rendered { 265 + assert_eq!(dimensions(&out.bytes), (variant.edge(), variant.edge())); 266 + } 267 + } 268 + 269 + #[test] 270 + fn non_image_bytes_fail_to_decode() { 271 + assert!(render(b"definitely not an image", Variant::Base).is_err()); 272 + } 273 + 274 + #[test] 275 + fn only_decodable_image_types_are_resizable() { 276 + assert!(is_resizable("image/jpeg")); 277 + assert!(is_resizable("image/png; charset=binary")); 278 + assert!(is_resizable("IMAGE/WEBP")); 279 + assert!(!is_resizable("image/svg+xml")); 280 + assert!(!is_resizable("video/mp4")); 281 + assert!(!is_resizable("application/octet-stream")); 282 + } 283 + }
+1
src/lib/mod.rs
··· 27 27 pub mod http; 28 28 pub mod hum_client; 29 29 pub mod hum_guard; 30 + pub mod image_variant; 30 31 pub mod klipy; 31 32 pub mod list_atproto_records; 32 33 pub mod map_tap_event;
+32 -10
src/lib/repo_endpoint.rs
··· 108 108 109 109 #[cfg(test)] 110 110 mod tests { 111 + use std::sync::{Mutex, MutexGuard}; 112 + 111 113 use super::*; 112 114 use rocket::tokio; 113 115 use sea_orm::{DatabaseBackend, MockDatabase}; 114 116 115 117 use crate::models::community_credentials as community_credentials_model; 116 118 119 + /// `PDS_LOC` is process-global, but cargo runs tests in parallel threads, so 120 + /// without this every test that touches it can have the var pulled out from 121 + /// under it by another one mid-assertion. 122 + static PDS_LOC_LOCK: Mutex<()> = Mutex::new(()); 123 + 124 + /// Holds [`PDS_LOC_LOCK`] and sets `PDS_LOC` for the duration of a test, 125 + /// clearing it again on drop, including when the test panics, which a 126 + /// trailing `remove_var` would skip. 127 + struct PdsLocGuard(#[allow(dead_code)] MutexGuard<'static, ()>); 128 + 129 + impl PdsLocGuard { 130 + fn set(value: &str) -> Self { 131 + let guard = PDS_LOC_LOCK.lock().unwrap_or_else(|e| e.into_inner()); 132 + // SAFETY: `PDS_LOC_LOCK` is held, so no other test reads or writes 133 + // the variable until this guard is dropped. 134 + unsafe { std::env::set_var("PDS_LOC", value) }; 135 + PdsLocGuard(guard) 136 + } 137 + } 138 + 139 + impl Drop for PdsLocGuard { 140 + fn drop(&mut self) { 141 + // SAFETY: the lock is still held until this guard finishes dropping. 142 + unsafe { std::env::remove_var("PDS_LOC") }; 143 + } 144 + } 145 + 117 146 #[test] 118 147 fn as_str_trims_trailing_slash() { 119 148 assert_eq!( ··· 128 157 129 158 #[test] 130 159 fn only_managed_rows_on_the_configured_pds_are_trusted() { 131 - // SAFETY: single-threaded test, and the var is restored below. 132 - unsafe { std::env::set_var("PDS_LOC", "http://localhost:3000") }; 160 + let _pds_loc = PdsLocGuard::set("http://localhost:3000"); 133 161 134 162 assert!(is_own_pds("http://localhost:3000", SOURCE_APPVIEW_MANAGED)); 135 163 // Trailing slash / casing differences are the same endpoint. ··· 138 166 assert!(!is_own_pds("http://localhost:3000", "byo")); 139 167 // A managed row for some other host (e.g. after PDS_LOC changed). 140 168 assert!(!is_own_pds("http://other.example", SOURCE_APPVIEW_MANAGED)); 141 - 142 - unsafe { std::env::remove_var("PDS_LOC") }; 143 169 } 144 170 145 171 fn credentials_row(pds_endpoint: &str, source: &str) -> community_credentials_model::Model { ··· 164 190 /// with no network access, which is the property local development needs. 165 191 #[tokio::test] 166 192 async fn resolve_prefers_a_managed_row_on_our_own_pds() { 167 - unsafe { std::env::set_var("PDS_LOC", "http://localhost:3000") }; 193 + let _pds_loc = PdsLocGuard::set("http://localhost:3000"); 168 194 169 195 let db = db_with_row(credentials_row( 170 196 "http://localhost:3000", ··· 176 202 endpoint, 177 203 RepoEndpoint::Trusted(String::from("http://localhost:3000")) 178 204 ); 179 - 180 - unsafe { std::env::remove_var("PDS_LOC") }; 181 205 } 182 206 183 207 /// A BYO endpoint is caller-supplied, so it stays behind the SSRF guard even 184 208 /// though we hold credentials for it. 185 209 #[tokio::test] 186 210 async fn resolve_keeps_byo_rows_untrusted() { 187 - unsafe { std::env::set_var("PDS_LOC", "http://localhost:3000") }; 211 + let _pds_loc = PdsLocGuard::set("http://localhost:3000"); 188 212 189 213 let db = db_with_row(credentials_row("http://localhost:3000", "byo")); 190 214 let endpoint = resolve(&db, "did:plc:comm").await.unwrap(); ··· 193 217 endpoint, 194 218 RepoEndpoint::Untrusted(String::from("http://localhost:3000")) 195 219 ); 196 - 197 - unsafe { std::env::remove_var("PDS_LOC") }; 198 220 } 199 221 200 222 #[tokio::test]
+241 -29
src/xrpc/com/atproto/sync/get_blob_handler.rs
··· 1 + use std::collections::HashMap; 1 2 use std::io::Cursor; 3 + use std::sync::{Arc, LazyLock, Mutex}; 2 4 3 5 use bytes::Bytes; 4 6 use rocket::http::{ContentType, Status}; 5 7 use rocket::request::Request; 6 8 use rocket::response::{Responder, Response}; 9 + use rocket::tokio::sync::Mutex as AsyncMutex; 7 10 use rocket::{State, get, response}; 8 11 9 12 use sea_orm::DatabaseConnection; ··· 11 14 use crate::lib::blob_cache::{BlobCache, CacheEntry}; 12 15 use crate::lib::embed_fetch; 13 16 use crate::lib::http::HTTP; 17 + use crate::lib::image_variant::{self, Variant}; 14 18 use crate::lib::range::{RangeResult, parse_range}; 15 19 use crate::lib::repo_endpoint::{self, RepoEndpoint}; 16 20 use crate::lib::responses::{ErrorBody, ErrorResponse}; ··· 89 93 } 90 94 } 91 95 96 + /// The cache key for one rendition of a blob: the content-addressed CID (which 97 + /// also dedupes the same blob across DIDs), plus the variant when the caller 98 + /// asked for a resized one. 99 + pub fn cache_key(cid: &str, variant: Option<Variant>) -> String { 100 + match variant { 101 + Some(variant) => format!("{cid}@{}", variant.as_str()), 102 + None => cid.to_string(), 103 + } 104 + } 105 + 106 + /// One in-flight fetch per CID. not per cache key, because a miss on any 107 + /// variant is satisfied by the same upstream fetch. Without this, a member list 108 + /// loading cold sends one request per avatar per rendition to the PDS at once. 109 + static BLOB_LOCKS: LazyLock<Mutex<HashMap<String, Arc<AsyncMutex<()>>>>> = 110 + LazyLock::new(|| Mutex::new(HashMap::new())); 111 + 112 + fn blob_lock(cid: &str) -> Arc<AsyncMutex<()>> { 113 + BLOB_LOCKS 114 + .lock() 115 + .unwrap() 116 + .entry(cid.to_string()) 117 + .or_insert_with(|| Arc::new(AsyncMutex::new(()))) 118 + .clone() 119 + } 120 + 121 + /// Drops `cid`'s lock once nobody is holding or waiting on it. There are far 122 + /// more distinct CIDs than DIDs, so the registry has to be pruned rather than 123 + /// grown forever. Losing a race here only costs a redundant fetch, never 124 + /// correctness. 125 + fn release_blob_lock(cid: &str) { 126 + let mut locks = BLOB_LOCKS.lock().unwrap(); 127 + if locks 128 + .get(cid) 129 + .is_some_and(|lock| Arc::strong_count(lock) == 1) 130 + { 131 + locks.remove(cid); 132 + } 133 + } 134 + 92 135 async fn get_blob_inner( 93 136 did: &str, 94 137 cid: &str, 138 + variant: Option<Variant>, 95 139 cache: &BlobCache, 96 140 db: &DatabaseConnection, 97 141 ) -> GetBlobResponse { 98 - // Cache hit: serve straight from memory (keyed by the content-addressed CID, 99 - // which also dedupes the same blob across DIDs). 100 - if let Some(entry) = cache.get(cid) { 142 + let key = cache_key(cid, variant); 143 + 144 + if let Some(entry) = cache.get(&key) { 145 + return GetBlobResponse::Blob { 146 + bytes: entry.bytes, 147 + content_type: entry.content_type, 148 + }; 149 + } 150 + 151 + let lock = blob_lock(cid); 152 + let response = { 153 + let _guard = lock.lock().await; 154 + fetch_and_cache(did, cid, variant, &key, cache, db).await 155 + }; 156 + drop(lock); 157 + release_blob_lock(cid); 158 + 159 + response 160 + } 161 + 162 + async fn fetch_and_cache( 163 + did: &str, 164 + cid: &str, 165 + variant: Option<Variant>, 166 + key: &str, 167 + cache: &BlobCache, 168 + db: &DatabaseConnection, 169 + ) -> GetBlobResponse { 170 + // Whoever held the lock before us may have already populated this key. 171 + if let Some(entry) = cache.get(key) { 101 172 return GetBlobResponse::Blob { 102 173 bytes: entry.bytes, 103 174 content_type: entry.content_type, 104 175 }; 105 176 } 106 177 178 + let (bytes, content_type) = match fetch_upstream(did, cid, db).await { 179 + Ok(fetched) => fetched, 180 + Err(response) => return response, 181 + }; 182 + 183 + // A variant request that we can actually decode: derive every size from the 184 + // one decode and cache those instead of the full-resolution original, so a 185 + // 250 KB avatar never occupies the cache (or the wire) to render a 40 px row. 186 + if variant.is_some() 187 + && image_variant::is_resizable(&content_type) 188 + && let Some(mut rendered) = render_variants(cid, bytes.clone(), cache).await 189 + && let Some(entry) = rendered.remove(key) 190 + { 191 + return GetBlobResponse::Blob { 192 + bytes: entry.bytes, 193 + content_type: entry.content_type, 194 + }; 195 + } 196 + 197 + cache.insert( 198 + &cache_key(cid, None), 199 + CacheEntry { 200 + bytes: bytes.clone(), 201 + content_type: content_type.clone(), 202 + }, 203 + ); 204 + 205 + GetBlobResponse::Blob { 206 + bytes, 207 + content_type, 208 + } 209 + } 210 + 211 + /// Decodes `bytes` once on a blocking worker, caches every variant, and returns 212 + /// them. `None` when the bytes turn out not to be a decodable image, in which 213 + /// case the caller falls back to serving the original. 214 + async fn render_variants( 215 + cid: &str, 216 + bytes: Bytes, 217 + cache: &BlobCache, 218 + ) -> Option<std::collections::HashMap<String, CacheEntry>> { 219 + let rendered = rocket::tokio::task::spawn_blocking(move || image_variant::render_all(&bytes)) 220 + .await 221 + .map_err(|e| e.to_string()) 222 + .and_then(|result| result.map_err(|e| e.to_string())); 223 + 224 + let rendered = match rendered { 225 + Ok(rendered) => rendered, 226 + Err(e) => { 227 + log::warn!("Failed to render image variants for {cid}: {e}"); 228 + return None; 229 + } 230 + }; 231 + 232 + let mut entries = std::collections::HashMap::new(); 233 + for (variant, out) in rendered { 234 + let entry = CacheEntry { 235 + bytes: out.bytes, 236 + content_type: out.content_type, 237 + }; 238 + let key = cache_key(cid, Some(variant)); 239 + cache.insert(&key, entry.clone()); 240 + entries.insert(key, entry); 241 + } 242 + 243 + Some(entries) 244 + } 245 + 246 + async fn fetch_upstream( 247 + did: &str, 248 + cid: &str, 249 + db: &DatabaseConnection, 250 + ) -> Result<(Bytes, String), GetBlobResponse> { 107 251 // For a community this AppView provisioned, the endpoint comes straight off 108 252 // its credentials row 109 253 let endpoint = match repo_endpoint::resolve(db, did).await { 110 254 Ok(e) => e, 111 255 Err(e) => { 112 - return GetBlobResponse::Upstream(ErrorResponse { 256 + return Err(GetBlobResponse::Upstream(ErrorResponse { 113 257 body: Json(ErrorBody { 114 258 error: String::from("UpstreamError"), 115 259 message: format!("Failed to resolve DID: {e}"), 116 260 }), 117 - }); 261 + })); 118 262 } 119 263 }; 120 264 ··· 129 273 .send() 130 274 .await 131 275 .map_err(|e| e.to_string()), 132 - RepoEndpoint::Untrusted(_) => embed_fetch::guarded_get(&url) 133 - .await 134 - .map_err(|e| e.to_string()), 276 + RepoEndpoint::Untrusted(_) => { 277 + embed_fetch::guarded_get_with_timeout(&url, embed_fetch::BLOB_FETCH_TIMEOUT) 278 + .await 279 + .map_err(|e| e.to_string()) 280 + } 135 281 }; 136 282 let resp = match fetched { 137 283 Ok(r) => r, 138 284 Err(e) => { 139 - return GetBlobResponse::Upstream(ErrorResponse { 285 + return Err(GetBlobResponse::Upstream(ErrorResponse { 140 286 body: Json(ErrorBody { 141 287 error: String::from("UpstreamError"), 142 288 message: format!("Blob fetch failed: {e}"), 143 289 }), 144 - }); 290 + })); 145 291 } 146 292 }; 147 293 148 294 if resp.status() == reqwest::StatusCode::NOT_FOUND { 149 - return GetBlobResponse::NotFound; 295 + return Err(GetBlobResponse::NotFound); 150 296 } 151 297 152 298 if !resp.status().is_success() { 153 299 let status = resp.status().as_u16(); 154 300 let body = resp.text().await.unwrap_or_default(); 155 - return GetBlobResponse::Upstream(ErrorResponse { 301 + return Err(GetBlobResponse::Upstream(ErrorResponse { 156 302 body: Json(ErrorBody { 157 303 error: String::from("UpstreamError"), 158 304 message: format!("PDS returned {status}: {body}"), 159 305 }), 160 - }); 306 + })); 161 307 } 162 308 163 309 let content_type = resp ··· 170 316 let bytes = match read_capped(resp, MAX_BLOB_BYTES).await { 171 317 Ok(b) => b, 172 318 Err(e) => { 173 - return GetBlobResponse::Upstream(ErrorResponse { 319 + return Err(GetBlobResponse::Upstream(ErrorResponse { 174 320 body: Json(ErrorBody { 175 321 error: String::from("UpstreamError"), 176 322 message: format!("Failed to read blob bytes: {e}"), 177 323 }), 178 - }); 324 + })); 179 325 } 180 326 }; 181 327 182 - cache.insert( 183 - cid, 184 - CacheEntry { 185 - bytes: bytes.clone(), 186 - content_type: content_type.clone(), 187 - }, 188 - ); 189 - 190 - GetBlobResponse::Blob { 191 - bytes, 192 - content_type, 193 - } 328 + Ok((bytes, content_type)) 194 329 } 195 330 196 - #[get("/xrpc/com.atproto.sync.getBlob?<did>&<cid>")] 331 + #[get("/xrpc/com.atproto.sync.getBlob?<did>&<cid>&<variant>")] 197 332 /// Proxies a blob fetch to the PDS that hosts the given DID, caching the bytes 198 333 /// in memory and serving HTTP Range requests itself (the PDS doesn't), so media 199 334 /// players can read duration up front and seek. 335 + /// 336 + /// `variant` (`small` / `base` / `large`) asks for a square, downscaled 337 + /// rendition instead of the original. 200 338 pub async fn get_blob( 201 339 did: &str, 202 340 cid: &str, 341 + variant: Option<&str>, 203 342 cache: &State<BlobCache>, 204 343 db: &State<DatabaseConnection>, 205 344 ) -> GetBlobResponse { 206 - get_blob_inner(did, cid, cache.inner(), db.inner()).await 345 + let variant = variant.and_then(Variant::parse); 346 + get_blob_inner(did, cid, variant, cache.inner(), db.inner()).await 347 + } 348 + 349 + #[cfg(test)] 350 + mod tests { 351 + use super::*; 352 + use rocket::tokio; 353 + 354 + const CID: &str = "bafkreiexamplecid"; 355 + 356 + #[test] 357 + fn variants_and_originals_get_distinct_cache_keys() { 358 + let original = cache_key(CID, None); 359 + let small = cache_key(CID, Some(Variant::Small)); 360 + let large = cache_key(CID, Some(Variant::Large)); 361 + 362 + assert_eq!(original, CID); 363 + assert_ne!(small, original); 364 + assert_ne!(small, large); 365 + } 366 + 367 + #[tokio::test] 368 + async fn a_held_lock_serializes_the_next_caller() { 369 + let cid = "bafkrei-serialized"; 370 + let first = blob_lock(cid); 371 + let guard = first.lock().await; 372 + 373 + let second = blob_lock(cid); 374 + assert!( 375 + second.try_lock().is_err(), 376 + "a second caller must wait rather than fetch upstream in parallel" 377 + ); 378 + 379 + drop(guard); 380 + assert!(second.try_lock().is_ok()); 381 + 382 + drop(first); 383 + drop(second); 384 + release_blob_lock(cid); 385 + } 386 + 387 + #[test] 388 + fn releasing_prunes_the_registry() { 389 + let cid = "bafkrei-pruned"; 390 + 391 + let lock = blob_lock(cid); 392 + assert!(BLOB_LOCKS.lock().unwrap().contains_key(cid)); 393 + 394 + drop(lock); 395 + release_blob_lock(cid); 396 + 397 + assert!( 398 + !BLOB_LOCKS.lock().unwrap().contains_key(cid), 399 + "the registry must not grow one entry per CID ever served" 400 + ); 401 + } 402 + 403 + #[test] 404 + fn releasing_keeps_a_lock_someone_else_still_holds() { 405 + let cid = "bafkrei-contended"; 406 + 407 + let held = blob_lock(cid); 408 + let mine = blob_lock(cid); 409 + 410 + drop(mine); 411 + release_blob_lock(cid); 412 + 413 + assert!(BLOB_LOCKS.lock().unwrap().contains_key(cid)); 414 + 415 + drop(held); 416 + release_blob_lock(cid); 417 + assert!(!BLOB_LOCKS.lock().unwrap().contains_key(cid)); 418 + } 207 419 }