-22

Cargo.lock

reviewed

··· 1868 1868 "jacquard-common", 1869 1869 "jacquard-lexicon", 1870 1870 "miette", 1871 1871 - "mini-moka-wasm", 1872 1871 "n0-future", 1873 1872 "percent-encoding", 1874 1873 "reqwest", ··· 2078 2077 version = "0.3.17" 2079 2078 source = "registry+https://github.com/rust-lang/crates.io-index" 2080 2079 checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" 2081 2081 - 2082 2082 - [[package]] 2083 2083 - name = "mini-moka-wasm" 2084 2084 - version = "0.10.99" 2085 2085 - source = "registry+https://github.com/rust-lang/crates.io-index" 2086 2086 - checksum = "0102b9a2ad50fa47ca89eead2316c8222285ecfbd3f69ce99564fbe4253866e8" 2087 2087 - dependencies = [ 2088 2088 - "crossbeam-channel", 2089 2089 - "crossbeam-utils", 2090 2090 - "dashmap", 2091 2091 - "smallvec", 2092 2092 - "tagptr", 2093 2093 - "triomphe", 2094 2094 - "web-time", 2095 2095 - ] 2096 2080 2097 2081 [[package]] 2098 2082 name = "minimal-lexical" ··· 3685 3669 "quote", 3686 3670 "syn 2.0.117", 3687 3671 ] 3688 3688 - 3689 3689 - [[package]] 3690 3690 - name = "triomphe" 3691 3691 - version = "0.1.15" 3692 3692 - source = "registry+https://github.com/rust-lang/crates.io-index" 3693 3693 - checksum = "dd69c5aa8f924c7519d6372789a74eac5b94fb0f8fcf0d4a97eb0bfc3e785f39" 3694 3672 3695 3673 [[package]] 3696 3674 name = "try-lock"

+1 -1

Cargo.toml

reviewed

··· 48 48 cid = "0.11.1" 49 49 multihash-codetable = { version = "0.1.4", features = ["sha2", "blake3"] } 50 50 jacquard-common = "0.9.5" 51 51 - jacquard-identity = { version = "0.9.5", features = ["cache"] } 51 51 + jacquard-identity = { version = "0.9.5" } 52 52 bytesize = { version = "2.3.1", features = ["serde"] } 53 53 mime = "0.3.17" 54 54 humantime = "2.3.0"

+66 -15

src/cache.rs

reviewed

··· 1 1 + use crate::types::blob_cid::BlobCid; 1 2 use anyhow::{Context, Result}; 2 3 use axum::http::HeaderMap; 3 4 use bytes::Bytes; 4 4 - use cid::Cid; 5 5 use jacquard_common::types::did::Did; 6 6 use moka::{future::Cache as MokaCache, policy::EvictionPolicy}; 7 7 - use std::{cmp, time::Duration}; 7 7 + use reqwest::Url; 8 8 + use std::{cmp, num::NonZeroU64, time::Duration}; 8 9 9 10 // Blob Content Cache 10 11 11 11 - type BlobContentCache = MokaCache<Cid, CachedBlobData>; 12 12 + type BlobContentCache = MokaCache<BlobCid, CachedBlobData>; 12 13 13 14 #[derive(Clone)] 14 15 pub struct CachedBlobData { ··· 16 17 pub headers: HeaderMap, 17 18 } 18 19 20 20 + #[must_use] 19 21 fn build_blob_content_cache(mem_capacity: u64, ttl: Duration) -> BlobContentCache { 20 22 tracing::debug!( 21 23 "building blob content cache with a mem_capacity of {mem_capacity} bytes and a ttl of {}s", ··· 24 26 25 27 BlobContentCache::builder() 26 28 .name("blob-content") 27 27 - .weigher(|_key, value: &CachedBlobData| -> u32 { 28 28 - (value.bytes.len() as u64 + value.headers.len() as u64 * 64) 29 29 - .try_into() 30 30 - .unwrap_or(u32::MAX) 29 29 + .weigher(|_key, value| { 30 30 + (value.bytes.len() 31 31 + + value 32 32 + .headers 33 33 + .iter() 34 34 + .map(|(k, v)| k.as_str().len() + v.len() + 32) 35 35 + .sum::<usize>()) 36 36 + .try_into() 37 37 + .unwrap_or(u32::MAX) 31 38 }) 32 39 .eviction_policy(EvictionPolicy::tiny_lfu()) 33 40 .max_capacity(mem_capacity) ··· 37 44 38 45 // Blob Ownership Cache 39 46 40 40 - type BlobOwnershipCache = MokaCache<(Cid, Did<'static>), ()>; 47 47 + type BlobOwnershipCache = MokaCache<(BlobCid, Did<'static>), ()>; 41 48 49 49 + #[must_use] 42 50 fn build_blob_ownership_cache(mem_capacity: u64, ttl: Duration) -> BlobOwnershipCache { 43 51 tracing::debug!( 44 52 "building blob ownership cache with a mem_capacity of {mem_capacity} bytes and a ttl of {}s", ··· 47 55 48 56 BlobOwnershipCache::builder() 49 57 .name("blob-ownership") 50 50 - .weigher(|key, _value| -> u32 { 58 58 + .weigher(|key, _value| { 51 59 (key.0.encoded_len() + key.1.len()) 52 60 .try_into() 53 61 .unwrap_or(u32::MAX) ··· 61 69 62 70 // Policy Cache 63 71 64 64 - type BlobPolicyCache = MokaCache<(Did<'static>, Cid), CachedBlobPolicy>; 72 72 + type BlobPolicyCache = MokaCache<(Did<'static>, BlobCid), CachedBlobPolicy>; 65 73 66 74 #[derive(Debug, Copy, Clone)] 67 75 pub struct CachedBlobPolicy { 68 76 pub can_serve: bool, 69 77 } 70 78 79 79 + #[must_use] 71 80 pub fn build_blob_policy_cache(mem_capacity: u64, ttl: Duration) -> BlobPolicyCache { 72 81 tracing::debug!( 73 82 "building blob policy cache with a mem_capacity of {mem_capacity} bytes and a ttl of {}s", ··· 76 85 77 86 BlobPolicyCache::builder() 78 87 .name("blob-policy") 79 79 - .weigher(|key, _value| -> u32 { 88 88 + .weigher(|key, _value| { 80 89 (key.0.len() + key.1.encoded_len()) 81 90 .try_into() 82 91 .unwrap_or(u32::MAX) ··· 88 97 .build() 89 98 } 90 99 100 100 + // Identity DID <-> PDS Cache 101 101 + 102 102 + type IdentityCache = MokaCache<Did<'static>, Url>; 103 103 + 104 104 + #[must_use] 105 105 + pub fn build_identity_cache(mem_capacity: u64, ttl: Duration) -> IdentityCache { 106 106 + tracing::debug!( 107 107 + "building identity cache with a mem_capacity of {mem_capacity} bytes and a ttl of {}s", 108 108 + ttl.as_secs() 109 109 + ); 110 110 + 111 111 + IdentityCache::builder() 112 112 + .name("identity") 113 113 + .weigher(|key, value| { 114 114 + (key.len() + value.as_str().len()) 115 115 + .try_into() 116 116 + .unwrap_or(u32::MAX) 117 117 + }) 118 118 + .eviction_policy(EvictionPolicy::tiny_lfu()) 119 119 + .max_capacity(mem_capacity) 120 120 + .time_to_live(ttl) 121 121 + .support_invalidation_closures() 122 122 + .build() 123 123 + } 124 124 + 91 125 // Builder 92 126 93 127 pub struct Caches { 94 128 pub blob_content: BlobContentCache, 95 129 pub blob_ownership: BlobOwnershipCache, 96 130 pub blob_policy: BlobPolicyCache, 131 131 + pub identity: IdentityCache, 97 132 } 98 133 99 134 pub struct CacheBuildOptions { 100 100 - pub memory_capacity: u64, 135 135 + pub memory_capacity: NonZeroU64, 101 136 pub blob_content_ttl: Duration, 102 137 pub blob_ownership_ttl: Duration, 103 138 pub blob_policy_ttl: Duration, 139 139 + pub identity_cache_ttl: Duration, 104 140 } 105 141 106 142 pub fn build_caches(options: &CacheBuildOptions) -> Result<Caches> { ··· 109 145 pub blob: u64, 110 146 pub ownership: u64, 111 147 pub policy: u64, 148 148 + pub identity: u64, 112 149 } 113 113 - let policy = cmp::min((options.memory_capacity as f64 * 0.10) as u64, 68_000_000); // 10% up to 68mb max (roughly. 1mil entries) 114 114 - let ownership = cmp::min((options.memory_capacity as f64 * 0.10) as u64, 68_000_000); // 10% up to 68mb max (roughly. 1mil entries) 150 150 + let policy = cmp::min( 151 151 + (options.memory_capacity.get() as f64 * 0.06) as u64, 152 152 + 48_000_000, 153 153 + ); // 6% up to 48mb max. 154 154 + let ownership = cmp::min( 155 155 + (options.memory_capacity.get() as f64 * 0.06) as u64, 156 156 + 48_000_000, 157 157 + ); // 6% up to 48mb max. 158 158 + let identity = cmp::min( 159 159 + (options.memory_capacity.get() as f64 * 0.06) as u64, 160 160 + 48_000_000, 161 161 + ); // 6% up to 48mb max. 115 162 CacheSizes { 116 163 policy, 117 164 ownership, 165 165 + identity, 118 166 blob: options 119 167 .memory_capacity 168 168 + .get() 120 169 .checked_sub(policy) 121 170 .and_then(|r| r.checked_sub(ownership)) 122 122 - .context("cache size allocation overflow")?, 171 171 + .and_then(|r| r.checked_sub(identity)) 172 172 + .context("cache size allocation underflow")?, 123 173 } 124 174 }; 125 175 ··· 127 177 blob_content: build_blob_content_cache(sizes.blob, options.blob_content_ttl), 128 178 blob_ownership: build_blob_ownership_cache(sizes.ownership, options.blob_ownership_ttl), 129 179 blob_policy: build_blob_policy_cache(sizes.policy, options.blob_policy_ttl), 180 180 + identity: build_identity_cache(sizes.identity, options.identity_cache_ttl), 130 181 }) 131 182 }

+2 -2

src/http.rs

reviewed

··· 37 37 /// The response content length exceeded the size limit. 38 38 #[error("content exceeded the maximum size")] 39 39 TooLarge, 40 40 - /// An internal client error occured whilst processing the request, 40 40 + /// An internal client error occurred whilst processing the request, 41 41 /// see [`reqwest::Error`]. 42 42 - #[error("an internal client error occured: {0}")] 42 42 + #[error(transparent)] 43 43 ClientError(#[from] reqwest::Error), 44 44 } 45 45

+22 -18

src/main.rs

reviewed

··· 2 2 mod http; 3 3 mod mime; 4 4 mod routes; 5 5 + mod types; 5 6 6 7 use crate::{ 7 8 cache::{CacheBuildOptions, Caches, build_caches}, ··· 9 10 routes::{delete_cache_handler, get_blob_handler, get_index_handler}, 10 11 }; 11 12 use ::mime::Mime; 12 12 - #[cfg(not(unix))] 13 13 - use anyhow::bail; 14 13 use anyhow::{Context, Result, bail}; 15 14 use axum::{ 16 15 Router, ··· 143 142 long = "blob-max-size", 144 143 env = "PORXIE_BLOB_MAX_SIZE", 145 144 default_value = "50mb", 146 146 - value_parser = |v: &str| -> Result<ByteSize, String> { 145 145 + value_parser = |v: &str| -> Result<NonZeroU64, String> { 147 146 let size: ByteSize = v.parse().map_err(|e| format!("{e}"))?; 148 147 if size.as_u64() < 512_000 { 149 148 return Err("minimum allowed value is 512kb".to_string()) ··· 159 158 )); 160 159 } 161 160 162 162 - Ok(size) 161 161 + Ok(size.as_u64().try_into().map_err(|e| format!("invalid value {v}: {e}"))?) 163 162 } 164 163 )] 165 165 - max_size: ByteSize, 164 164 + max_size: NonZeroU64, 166 165 167 166 /// The Cache-Control header value to send alongside blob responses. 168 167 /// ··· 259 258 long = "cache-allocation", 260 259 env = "PORXIE_CACHE_ALLOCATION", 261 260 default_value = "512mb", 262 262 - value_parser = |v: &str| -> Result<ByteSize, String> { 261 261 + value_parser = |v: &str| -> Result<NonZeroU64, String> { 263 262 let size: ByteSize = v.parse().map_err(|e| format!("{e}"))?; 264 263 if size.as_u64() < 8_000_000 { 265 264 return Err("minimum allowed value is 8mb".to_string()) ··· 275 274 )); 276 275 } 277 276 278 278 - Ok(size) 277 277 + Ok(size.as_u64().try_into().map_err(|e| format!("invalid value {v}: {e}"))?) 279 278 } 280 279 )] 281 281 - size: ByteSize, 280 280 + size: NonZeroU64, 282 281 283 282 /// How long blobs can be idle in the cache before expiring. 284 283 #[arg( ··· 306 305 default_value = "1h" 307 306 )] 308 307 policy_ttl: humantime::Duration, 308 308 + 309 309 + /// How long identity lookups (DID resolution, etc) can be cached before expiring. 310 310 + #[arg( 311 311 + id = "CA_CACHE_IDENTITY_TTL", 312 312 + long = "cache-identity-ttl", 313 313 + env = "PORXIE_CACHE_IDENTITY_TTL", 314 314 + default_value = "1h" 315 315 + )] 316 316 + identity_ttl: humantime::Duration, 309 317 } 310 318 311 319 #[derive(Args)] ··· 337 345 requires = "PA_POLICY_URL", 338 346 value_parser = |v: &str| -> Result<(HeaderName, HeaderValue), String> { 339 347 let (name, value) = v.split_once(':') 340 340 - .ok_or_else(|| format!("invalid header {v:?}: expected 'Name: value'"))?; 348 348 + .ok_or_else(|| format!("invalid header {v}: expected 'Name: value'"))?; 341 349 let name = HeaderName::try_from(name.trim()) 342 342 - .map_err(|e| format!("invalid header name in {v:?}: {e}"))?; 350 350 + .map_err(|e| format!("invalid header name in {v}: {e}"))?; 343 351 let mut value = HeaderValue::try_from(value.trim()) 344 344 - .map_err(|e| format!("invalid header value in {v:?}: {e}"))?; 352 352 + .map_err(|e| format!("invalid header value in {v}: {e}"))?; 345 353 value.set_sensitive(true); 346 354 Ok((name, value)) 347 355 } ··· 439 447 ) 440 448 .context("failed to build policy http client")?, 441 449 cache: build_caches(&CacheBuildOptions { 442 442 - memory_capacity: args.cache.size.as_u64(), 450 450 + memory_capacity: args.cache.size, 443 451 blob_content_ttl: args.cache.blob_tti.into(), 444 452 blob_ownership_ttl: args.cache.ownership_ttl.into(), 445 453 blob_policy_ttl: args.cache.policy_ttl.into(), 454 454 + identity_cache_ttl: args.cache.identity_ttl.into(), 446 455 }) 447 456 .context("failed to build caches")?, 448 457 auth_token: args.server.auth_token, 449 458 allowed_mimetypes: args.blob.allowed_mimetypes, 450 450 - max_blob_size: args 451 451 - .blob 452 452 - .max_size 453 453 - .as_u64() 454 454 - .try_into() 455 455 - .context("max blob size was not a non-zero value")?, 459 459 + max_blob_size: args.blob.max_size, 456 460 cache_control_header: args.blob.cache_header, 457 461 policy_service_url: args.policy.url, 458 462 policy_service_headers: args.policy.request_headers,

+9 -22

src/mime.rs

reviewed

··· 1 1 use mime::Mime; 2 2 3 3 /// Sniff the MIME type from the given bytes, returning `application/octet-stream` if unknown. 4 4 + #[must_use] 4 5 pub fn sniff_mime(buf: &[u8]) -> Mime { 5 6 // WORKAROUND: infer does not correctly detect SVG. 6 7 // I have created PR to fix this at https://github.com/bojand/infer/pull/119 ··· 28 29 } 29 30 } 30 31 32 32 + #[must_use] 31 33 pub fn is_mime_allowed(mime: &Mime, allowed: &[Mime]) -> bool { 32 34 const STAR: &str = "*"; 33 35 ··· 64 66 fn test_is_mime_allowed() { 65 67 // Test PNG when nothing is allowed. 66 68 assert_eq!( 67 67 - super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &vec![]), 69 69 + super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &[]), 68 70 false 69 71 ); 70 72 71 73 // Test PNG when PNG is allowed. 72 74 assert_eq!( 73 73 - super::is_mime_allowed( 74 74 - &Mime::from_str("image/png").unwrap(), 75 75 - &vec![mime::IMAGE_PNG], 76 76 - ), 75 75 + super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &[mime::IMAGE_PNG],), 77 76 true 78 77 ); 79 78 80 79 // Test PNG when only JPG is allowed. 81 80 assert_eq!( 82 82 - super::is_mime_allowed( 83 83 - &Mime::from_str("image/png").unwrap(), 84 84 - &vec![mime::IMAGE_JPEG], 85 85 - ), 81 81 + super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &[mime::IMAGE_JPEG],), 86 82 false 87 83 ); 88 84 89 85 // Test PNG when any image subtype is allowed. 90 86 assert_eq!( 91 91 - super::is_mime_allowed( 92 92 - &Mime::from_str("image/png").unwrap(), 93 93 - &vec![mime::IMAGE_STAR], 94 94 - ), 87 87 + super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &[mime::IMAGE_STAR],), 95 88 true 96 89 ); 97 90 98 91 // Test PNG when anything is allowed. 99 92 assert_eq!( 100 100 - super::is_mime_allowed( 101 101 - &Mime::from_str("image/png").unwrap(), 102 102 - &vec![mime::STAR_STAR], 103 103 - ), 93 93 + super::is_mime_allowed(&Mime::from_str("image/png").unwrap(), &[mime::STAR_STAR],), 104 94 true 105 95 ); 106 96 107 97 // Test HTML when any image subtype is enabled. 108 98 assert_eq!( 109 109 - super::is_mime_allowed( 110 110 - &Mime::from_str("text/html").unwrap(), 111 111 - &vec![mime::IMAGE_STAR], 112 112 - ), 99 99 + super::is_mime_allowed(&Mime::from_str("text/html").unwrap(), &[mime::IMAGE_STAR],), 113 100 false 114 101 ); 115 102 ··· 117 104 assert_eq!( 118 105 super::is_mime_allowed( 119 106 &Mime::from_str("image/png").unwrap(), 120 120 - &vec![mime::TEXT_STAR, mime::IMAGE_STAR], 107 107 + &[mime::TEXT_STAR, mime::IMAGE_STAR], 121 108 ), 122 109 true 123 110 );

+48 -32

src/routes/blob/get.rs

reviewed

··· 1 1 use crate::http::{BytesStreamCappedError, bytes_stream_capped}; 2 2 use crate::routes::ErrorResponse; 3 3 + use crate::types::blob_cid::BlobCid; 3 4 use crate::{ 4 5 AppState, 5 6 cache::{CachedBlobData, CachedBlobPolicy}, ··· 13 14 }; 14 15 use cid::Cid; 15 16 use jacquard_common::types::did::Did; 16 16 - use jacquard_identity::resolver::{IdentityError, IdentityResolver}; 17 17 + use jacquard_identity::resolver::IdentityResolver; 17 18 use multihash_codetable::{Code, MultihashDigest}; 18 19 use reqwest::Url; 19 20 use std::sync::Arc; 20 21 21 22 enum BlobPolicyError { 22 22 - /// The policy service returned an unexpected status code,. 23 23 + /// The policy service returned an unexpected status code. 23 24 UnhandledStatusCode, 24 25 /// The request to the policy service failed, for example due to the server being unavailable. 25 26 FetchFailed, ··· 62 63 FetchFailure, 63 64 } 64 65 66 66 + /// Create a `/xrpc/com.atproto.sync.getBlob` Url for the DID+CID. 67 67 + #[inline] 68 68 + #[must_use] 69 69 + fn to_pds_blob_url(mut pds_url: Url, did: &Did<'_>, cid: &BlobCid) -> Url { 70 70 + pds_url.set_path("/xrpc/com.atproto.sync.getBlob"); 71 71 + pds_url 72 72 + .query_pairs_mut() 73 73 + .append_pair("did", did.as_str()) 74 74 + .append_pair("cid", &cid.to_string()); 75 75 + pds_url 76 76 + } 77 77 + 65 78 pub async fn get_blob_handler( 66 79 Path((raw_did, raw_cid)): Path<(String, String)>, 67 80 State(state): State<Arc<AppState>>, 68 81 ) -> Result<axum::response::Response, (StatusCode, Json<ErrorResponse>)> { 69 69 - /// Resolve the given DID to a PDS URL and then build the `/xrpc/com.atproto.sync.getBlob` url for the DID+CID. 70 70 - #[inline] 71 71 - async fn get_blob_url( 72 72 - state: &AppState, 73 73 - did: &Did<'_>, 74 74 - cid: &Cid, 75 75 - ) -> Result<Url, IdentityError> { 76 76 - let mut url = state.identity_resolver.pds_for_did(did).await?; 77 77 - url.set_path("/xrpc/com.atproto.sync.getBlob"); 78 78 - url.query_pairs_mut() 79 79 - .append_pair("did", did.as_str()) 80 80 - .append_pair("cid", &cid.to_string()); 81 81 - Ok(url) 82 82 - } 83 83 - 84 82 let (did, cid) = ( 85 83 match Did::new_owned(raw_did.as_str()) { 86 84 Ok(did) => did, ··· 94 92 )); 95 93 } 96 94 }, 97 97 - match Cid::try_from(raw_cid.as_str()) { 95 95 + match BlobCid::try_from(raw_cid.as_str()) { 98 96 Ok(cid) => cid, 99 97 Err(_) => { 100 98 return Err(( ··· 108 106 }, 109 107 ); 110 108 111 111 - // Check policy for this DID+CID; concurrent requests for a key are coalesced. 109 109 + // Check policy for this DID+CID; concurrent requests for the same key are coalesced. 112 110 if let Some(ref policy_service_url) = state.policy_service_url { 113 111 match state 114 112 .cache ··· 186 184 .blob_content 187 185 .try_get_with_by_ref(&cid, async { 188 186 tracing::debug!("fetching blob from PDS"); 189 189 - let blob_url = get_blob_url(&state, &did, &cid).await.map_err(|err| { 190 190 - tracing::debug!("failed to resolve PDS: {err:?}"); 191 191 - BlobDownloadError::DidPdsResolutionFailure 192 192 - })?; 187 187 + let blob_url = to_pds_blob_url( 188 188 + state 189 189 + .cache 190 190 + .identity 191 191 + .try_get_with_by_ref(&did, state.identity_resolver.pds_for_did(&did)) 192 192 + .await 193 193 + .map_err(|err| { 194 194 + tracing::debug!("failed to resolve PDS: {:?}", *err); 195 195 + BlobDownloadError::DidPdsResolutionFailure 196 196 + })?, 197 197 + &did, 198 198 + &cid, 199 199 + ); 193 200 194 201 let validated_bytes = { 195 202 let response = state ··· 204 211 205 212 // Gracefully handle & abort if we do not receive a successful status code. 206 213 if !response.status().is_success() { 207 207 - // Note: Bluesky's PDS implemenation sends 400 instead of 404 when a blob is 214 214 + // Note: Bluesky's PDS implementation sends 400 instead of 404 when a blob is 208 215 // not found. This will skip the 404 handler and instead count as an error. 209 216 // This is not our responsibility to work around as other implementations do it right. 210 217 return Err(match response.status() { ··· 253 260 } 254 261 }?; 255 262 256 256 - if computed_cid != cid { 263 263 + if computed_cid != *cid { 257 264 tracing::warn!("cid mismatch: computed {computed_cid} expected {cid}"); 258 265 return Err(BlobDownloadError::CidMismatch); 259 266 } ··· 298 305 const { HeaderValue::from_static("attachment") }, 299 306 ); 300 307 301 301 - // Mark this key as verified in the the ownership cache. 308 308 + // Mark this key as verified in the ownership cache. 302 309 state 303 310 .cache 304 311 .blob_ownership ··· 377 384 .blob_ownership 378 385 .try_get_with((cid, did.clone()), async { 379 386 tracing::debug!("verifying ownership of blob"); 380 380 - let blob_url = get_blob_url(&state, &did, &cid).await.map_err(|err| { 381 381 - tracing::debug!("failed to resolve PDS url: {err:?}"); 382 382 - BlobOwnershipError::DidPdsResolutionFailure 383 383 - })?; 387 387 + let blob_url = to_pds_blob_url( 388 388 + state 389 389 + .cache 390 390 + .identity 391 391 + .try_get_with_by_ref(&did, state.identity_resolver.pds_for_did(&did)) 392 392 + .await 393 393 + .map_err(|err| { 394 394 + tracing::debug!("failed to resolve PDS: {:?}", *err); 395 395 + BlobOwnershipError::DidPdsResolutionFailure 396 396 + })?, 397 397 + &did, 398 398 + &cid, 399 399 + ); 384 400 385 401 // Request the blob with as little of the actual body as we can. 386 402 // 387 403 // While some PDS implementations (bsky, tranquil) support HTTP HEAD, it is not 388 388 - // actually apart of the XRPC specification and we cannot rely on it (for now). 404 404 + // actually a part of the XRPC specification and we cannot rely on it (for now). 389 405 // Use a range request to avoid downloading the full body on servers that support it instead. 390 406 match state 391 407 .blob_fetch_http_client 392 408 .get(blob_url) 393 409 .header( 394 410 header::RANGE, 395 395 - const { HeaderValue::from_static("bytes=0-1") }, 411 411 + const { HeaderValue::from_static("bytes=0-1023") }, 396 412 ) 397 413 .send() 398 414 .await

+24 -8

src/routes/cache/delete.rs

reviewed

··· 1 1 - use crate::{AppState, routes::ErrorResponse}; 1 1 + use crate::{AppState, routes::ErrorResponse, types::blob_cid::BlobCid}; 2 2 use axum::{ 3 3 Json, 4 4 extract::{Path, State}, ··· 8 8 TypedHeader, 9 9 headers::{Authorization, authorization::Bearer}, 10 10 }; 11 11 - use cid::Cid; 12 11 use jacquard_common::types::did::Did; 13 12 use std::sync::Arc; 14 13 ··· 39 38 ) 40 39 })?; 41 40 42 42 - // Clear all ownership and policy data for this DID. 41 41 + // Clear all identity, ownership and policy data for this DID. 42 42 + state 43 43 + .cache 44 44 + .identity 45 45 + .invalidate_entries_if({ 46 46 + let did = did.clone(); 47 47 + move |k, _v| *k == did 48 48 + }) 49 49 + .map_err(|err| { 50 50 + tracing::error!("failed to schedule identity cache invalidation: {err:?}"); 51 51 + ( 52 52 + StatusCode::INTERNAL_SERVER_ERROR, 53 53 + Json(ErrorResponse { 54 54 + error: "InternalServerError", 55 55 + message: Some("Failed to schedule cache invalidation"), 56 56 + }), 57 57 + ) 58 58 + })?; 43 59 state 44 60 .cache 45 61 .blob_policy ··· 48 64 move |k, _v| k.0 == did 49 65 }) 50 66 .map_err(|err| { 51 51 - tracing::error!("failed to invalid entries: {err:?}"); 67 67 + tracing::error!("failed to schedule blob policy cache invalidation: {err:?}"); 52 68 ( 53 69 StatusCode::INTERNAL_SERVER_ERROR, 54 70 Json(ErrorResponse { ··· 62 78 .blob_ownership 63 79 .invalidate_entries_if(move |k, _v| k.1 == did) 64 80 .map_err(|err| { 65 65 - tracing::error!("failed to invalid entries: {err:?}"); 81 81 + tracing::error!("failed to schedule blob ownership cache invalidation: {err:?}"); 66 82 ( 67 83 StatusCode::INTERNAL_SERVER_ERROR, 68 84 Json(ErrorResponse { ··· 73 89 })?; 74 90 } else { 75 91 tracing::info!("invalidating CID cache entries"); 76 76 - let cid = Cid::try_from(identifier.as_str()).map_err(|_| { 92 92 + let cid = BlobCid::try_from(identifier.as_str()).map_err(|_| { 77 93 ( 78 94 StatusCode::UNPROCESSABLE_ENTITY, 79 95 Json(ErrorResponse { ··· 90 106 .blob_ownership 91 107 .invalidate_entries_if(move |k, _v| k.0 == cid) 92 108 .map_err(|err| { 93 93 - tracing::error!("failed to invalid entries: {err:?}"); 109 109 + tracing::error!("failed to schedule blob ownership cache invalidation: {err:?}"); 94 110 ( 95 111 StatusCode::INTERNAL_SERVER_ERROR, 96 112 Json(ErrorResponse { ··· 104 120 .blob_policy 105 121 .invalidate_entries_if(move |k, _v| k.1 == cid) 106 122 .map_err(|err| { 107 107 - tracing::error!("failed to invalid entries: {err:?}"); 123 123 + tracing::error!("failed to schedule blob policy cache invalidation: {err:?}"); 108 124 ( 109 125 StatusCode::INTERNAL_SERVER_ERROR, 110 126 Json(ErrorResponse {

+89

src/types/blob_cid.rs

reviewed

··· 1 1 + // TODO: Transfer this implementation to a standalone ATProto types crate in the future. 2 2 + 3 3 + use serde::Serialize; 4 4 + use thiserror::Error; 5 5 + 6 6 + pub mod codecs { 7 7 + pub const RAW: u64 = 0x55; 8 8 + } 9 9 + 10 10 + #[derive(Debug, Error)] 11 11 + pub enum BlobCidError { 12 12 + /// The CID uses a codec other than raw (`0x55`), which is the only codec 13 13 + /// permitted for ATProto blobs. 14 14 + #[error("invalid blob codec 0x{0:x}, the only supported codec is raw (0x55)")] 15 15 + InvalidBlobCodec(u64), 16 16 + 17 17 + /// The underlying CID could not be parsed. 18 18 + #[error(transparent)] 19 19 + CidError(#[from] cid::Error), 20 20 + } 21 21 + 22 22 + /// A [`cid::Cid`] wrapper that guarantees the codec is raw (`0x55`), conforming 23 23 + /// to the ATProto blob CID specification. 24 24 + /// 25 25 + /// Specification: <https://atproto.com/specs/blob> (Conformant as of **13/03/26**). 26 26 + #[derive(Copy, PartialEq, Eq, Clone, PartialOrd, Ord, Hash, Debug, Serialize)] 27 27 + pub struct BlobCid(cid::Cid); 28 28 + 29 29 + impl BlobCid { 30 30 + pub fn new(cid: cid::Cid) -> Result<Self, BlobCidError> { 31 31 + if cid.codec() != codecs::RAW { 32 32 + return Err(BlobCidError::InvalidBlobCodec(cid.codec())); 33 33 + } 34 34 + Ok(Self(cid)) 35 35 + } 36 36 + } 37 37 + 38 38 + impl<'de> serde::Deserialize<'de> for BlobCid { 39 39 + fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> { 40 40 + let cid = cid::Cid::deserialize(deserializer)?; 41 41 + Self::new(cid).map_err(serde::de::Error::custom) 42 42 + } 43 43 + } 44 44 + 45 45 + impl core::convert::TryFrom<&str> for BlobCid { 46 46 + type Error = BlobCidError; 47 47 + fn try_from(value: &str) -> Result<Self, Self::Error> { 48 48 + Self::new(cid::Cid::try_from(value)?) 49 49 + } 50 50 + } 51 51 + 52 52 + impl core::convert::TryFrom<String> for BlobCid { 53 53 + type Error = BlobCidError; 54 54 + fn try_from(value: String) -> Result<Self, Self::Error> { 55 55 + Self::new(cid::Cid::try_from(value)?) 56 56 + } 57 57 + } 58 58 + 59 59 + impl core::convert::TryFrom<Vec<u8>> for BlobCid { 60 60 + type Error = BlobCidError; 61 61 + fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> { 62 62 + Self::new(cid::Cid::try_from(value)?) 63 63 + } 64 64 + } 65 65 + 66 66 + impl core::convert::AsRef<cid::Cid> for BlobCid { 67 67 + fn as_ref(&self) -> &cid::Cid { 68 68 + &self.0 69 69 + } 70 70 + } 71 71 + 72 72 + impl core::fmt::Display for BlobCid { 73 73 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 74 74 + self.0.fmt(f) 75 75 + } 76 76 + } 77 77 + 78 78 + impl core::ops::Deref for BlobCid { 79 79 + type Target = cid::Cid; 80 80 + fn deref(&self) -> &Self::Target { 81 81 + &self.0 82 82 + } 83 83 + } 84 84 + 85 85 + impl core::borrow::Borrow<cid::Cid> for BlobCid { 86 86 + fn borrow(&self) -> &cid::Cid { 87 87 + &self.0 88 88 + } 89 89 + }

+2

src/types/mod.rs

reviewed

··· 1 1 + pub mod blob_cid; 2 2 + // pub mod validated_blob;s

+80

src/types/validated_blob.rs

reviewed

··· 1 1 + // // TODO: Consider transferring this implementation to a standalone ATProto crate in the future. 2 2 + 3 3 + // use crate::types::blob_cid::{self}; 4 4 + // use multihash_codetable::{Code, MultihashDigest}; 5 5 + // use thiserror::Error; 6 6 + 7 7 + // #[derive(Debug, Error)] 8 8 + // pub enum ValidatedBlobError { 9 9 + // /// The CID's multihash codec is not supported by the codetable. 10 10 + // #[error("unsupported multihash codec 0x{0:x}")] 11 11 + // CidUnsupportedMultihash(u64), 12 12 + // /// The computed CID of the blob content does not match the expected CID. 13 13 + // #[error("CID mismatch: computed {computed} but expected {expected}")] 14 14 + // CidMismatch { 15 15 + // computed: blob_cid::BlobCid, 16 16 + // expected: blob_cid::BlobCid, 17 17 + // }, 18 18 + // } 19 19 + 20 20 + // /// Blob content whose integrity has been verified against a [`blob_cid::BlobCid`]. 21 21 + // #[derive(Debug, PartialEq, Eq, Clone, PartialOrd, Ord, Hash)] 22 22 + // pub struct ValidatedBlob(bytes::Bytes); 23 23 + 24 24 + // impl ValidatedBlob { 25 25 + // /// Verify that `bytes` matches the expected `checksum` CID. 26 26 + // pub fn new<B: Into<bytes::Bytes>>( 27 27 + // bytes: B, 28 28 + // checksum: blob_cid::BlobCid, 29 29 + // ) -> Result<Self, ValidatedBlobError> { 30 30 + // let bytes = bytes.into(); 31 31 + 32 32 + // // Enabled Multihashes are set in the multihash-codetable crate features. 33 33 + // let hash_code = checksum.hash().code(); 34 34 + // let computed_cid = match Code::try_from(hash_code) { 35 35 + // Ok(code) => Ok(blob_cid::BlobCid::new(cid::Cid::new_v1( 36 36 + // blob_cid::codecs::RAW, 37 37 + // code.digest(&bytes), 38 38 + // )) 39 39 + // .expect("computed CID with raw codec should always be a valid BlobCid")), 40 40 + // Err(err) => { 41 41 + // tracing::warn!("failed to compute CID: {err:?}"); 42 42 + // Err(ValidatedBlobError::CidUnsupportedMultihash(hash_code)) 43 43 + // } 44 44 + // }?; 45 45 + 46 46 + // if computed_cid != checksum { 47 47 + // tracing::warn!("cid mismatch: computed {computed_cid} expected {checksum}"); 48 48 + // return Err(ValidatedBlobError::CidMismatch { 49 49 + // computed: computed_cid, 50 50 + // expected: checksum, 51 51 + // }); 52 52 + // } 53 53 + 54 54 + // Ok(Self(bytes)) 55 55 + // } 56 56 + 57 57 + // #[must_use] 58 58 + // pub fn into_inner(self) -> bytes::Bytes { 59 59 + // self.0 60 60 + // } 61 61 + // } 62 62 + 63 63 + // impl core::convert::AsRef<bytes::Bytes> for ValidatedBlob { 64 64 + // fn as_ref(&self) -> &bytes::Bytes { 65 65 + // &self.0 66 66 + // } 67 67 + // } 68 68 + 69 69 + // impl core::ops::Deref for ValidatedBlob { 70 70 + // type Target = bytes::Bytes; 71 71 + // fn deref(&self) -> &Self::Target { 72 72 + // &self.0 73 73 + // } 74 74 + // } 75 75 + 76 76 + // impl core::borrow::Borrow<bytes::Bytes> for ValidatedBlob { 77 77 + // fn borrow(&self) -> &bytes::Bytes { 78 78 + // &self.0 79 79 + // } 80 80 + // }