+11

bobbin/crates/bobbin/src/config.rs

View file Reviewed

··· 29 29 "search.heap_bytes", 30 30 "knot.allow_private", 31 31 "knot.require_https", 32 32 + "mirror.url", 32 33 "log.format", 33 34 "log.filter", 34 35 ]; ··· 52 53 "BOBBIN_SEARCH_HEAP_BYTES", 53 54 "BOBBIN_KNOT_ALLOW_PRIVATE", 54 55 "BOBBIN_KNOT_REQUIRE_HTTPS", 56 56 + "BOBBIN_MIRROR_URL", 55 57 "BOBBIN_LOG_FORMAT", 56 58 "BOBBIN_LOG", 57 59 ]; ··· 81 83 82 84 #[config(nested)] 83 85 pub knot: KnotConfig, 86 86 + 87 87 + #[config(nested)] 88 88 + pub mirror: MirrorConfig, 84 89 85 90 #[config(nested)] 86 91 pub log: LogConfig, ··· 251 256 /// knot for development. 252 257 #[config(env = "BOBBIN_KNOT_REQUIRE_HTTPS", default = true)] 253 258 pub require_https: bool, 259 259 + } 260 260 + 261 261 + #[derive(Debug, Config)] 262 262 + pub struct MirrorConfig { 263 263 + #[config(env = "BOBBIN_MIRROR_URL")] 264 264 + pub url: Option<Url>, 254 265 } 255 266 256 267 #[derive(Debug, Config)]

+17 -2

bobbin/crates/bobbin/src/main.rs

View file Reviewed

··· 11 11 IngestConfig, IngestRuntime, RepoIdResolver, WarmingBuffer, run as run_ingest, 12 12 }; 13 13 use bobbin_knot_ingest::{CapabilityGate, KnotClient, KnotRegistry, Orchestrator}; 14 14 - use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, classify_ip}; 14 14 + use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, MirrorProxy, classify_ip}; 15 15 use bobbin_record_lru::{CacheCapacity, LruRecordStore, RecordStore}; 16 16 use bobbin_runtime::{ 17 17 Clock, GuardedWs, MemoryBudget, NetworkError, OsEntropy, RuntimeHasher, SystemClock, ··· 201 201 }, 202 202 KnotHttpConfig::default(), 203 203 clock.clone(), 204 204 - hasher, 204 204 + hasher.clone(), 205 205 )?); 206 206 + let mirror = cfg 207 207 + .mirror 208 208 + .url 209 209 + .as_ref() 210 210 + .map(|url| MirrorProxy::new(url, clock.clone(), hasher.clone()).map(Arc::new)) 211 211 + .transpose() 212 212 + .context("mirror.url")?; 213 213 + match mirror.as_ref() { 214 214 + Some(m) => tracing::info!( 215 215 + mirror = %m.host().url(), 216 216 + "we will forward git reads to the mirror before any knot", 217 217 + ), 218 218 + None => tracing::info!("we will forward git reads to knots, since mirror.url is unset"), 219 219 + } 206 220 let search_heap = usize::try_from(search_heap_cap) 207 221 .with_context(|| format!("search heap {search_heap_cap} exceeds usize"))?; 208 222 let search = Arc::new(SearchIndex::new(search_heap, clock.clone())?); ··· 323 337 resolver, 324 338 ) 325 339 .with_limiter(limiter) 340 340 + .with_mirror(mirror) 326 341 .with_proxies(trusted_proxies); 327 342 let app = router(state); 328 343

+96 -28

bobbin/crates/xrpc/src/lib.rs

View file Reviewed

··· 14 14 HeaderMap, HeaderName, StatusCode, 15 15 header::{ 16 16 ACCEPT_RANGES, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LENGTH, 17 17 - CONTENT_RANGE, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, IF_NONE_MATCH, IF_RANGE, 18 18 - LAST_MODIFIED, RANGE, 17 17 + CONTENT_RANGE, CONTENT_SECURITY_POLICY, CONTENT_TYPE, ETAG, IF_MODIFIED_SINCE, 18 18 + IF_NONE_MATCH, IF_RANGE, LAST_MODIFIED, RANGE, X_CONTENT_TYPE_OPTIONS, 19 19 }, 20 20 request::Parts, 21 21 }, ··· 26 26 Coverage, CoverageWatch, CursorParseError, EdgeItem, EdgePage, EdgeStore, IssueStateKind, 27 27 PageCursor, PageLimit, PageToken, PullStatusKind, SortDir, StateIndex, StateKind, 28 28 }; 29 29 - use bobbin_knot_proxy::{KnotHost, KnotProxy, KnotProxyError, ProxyResponse, RepoSlug}; 29 29 + use bobbin_knot_proxy::{ 30 30 + KnotHost, KnotProxy, KnotProxyError, MirrorNsid, MirrorProxy, ProxyResponse, RepoSlug, 31 31 + }; 30 32 use bobbin_record_lru::RecordStore; 31 33 use bobbin_resolver::RepoIdResolver; 32 34 use bobbin_search::{ ··· 118 120 pub pull_statuses: Arc<StateIndex<PullStatusKind>>, 119 121 pub coverage: Arc<CoverageWatch>, 120 122 pub knots: Arc<KnotProxy>, 123 123 + pub mirror: Option<Arc<MirrorProxy>>, 121 124 pub search: Arc<dyn SearchReader>, 122 125 pub resolver: Arc<RepoIdResolver>, 123 126 pub limiter: Option<Arc<HeavyLimiter>>, ··· 145 148 pull_statuses, 146 149 coverage, 147 150 knots, 151 151 + mirror: None, 148 152 search, 149 153 resolver, 150 154 limiter: None, ··· 154 158 155 159 pub fn with_limiter(mut self, limiter: Option<Arc<HeavyLimiter>>) -> Self { 156 160 self.limiter = limiter; 161 161 + self 162 162 + } 163 163 + 164 164 + pub fn with_mirror(mut self, mirror: Option<Arc<MirrorProxy>>) -> Self { 165 165 + self.mirror = mirror; 157 166 self 158 167 } 159 168 ··· 468 477 &CONTENT_DISPOSITION, 469 478 &ACCEPT_RANGES, 470 479 &CONTENT_RANGE, 480 480 + &X_CONTENT_TYPE_OPTIONS, 481 481 + &CONTENT_SECURITY_POLICY, 471 482 ]; 472 483 473 473 - const FORWARDED_REQUEST_HEADERS: &[&HeaderName] = 484 484 + const RANGE_OR_CONDITIONAL_HEADERS: &[&HeaderName] = 474 485 &[&RANGE, &IF_RANGE, &IF_NONE_MATCH, &IF_MODIFIED_SINCE]; 475 486 476 487 const KNOT_HOST_PARAM: &str = "knot"; ··· 2637 2648 } 2638 2649 } 2639 2650 2651 2651 + struct RepoTarget { 2652 2652 + host: KnotHost, 2653 2653 + slug: RepoSlug, 2654 2654 + repo_did: Option<Did<DefaultStr>>, 2655 2655 + } 2656 2656 + 2640 2657 async fn resolve_knot_target( 2641 2658 state: &AppState, 2642 2659 repo_uri: AtUri<DefaultStr>, 2643 2643 - ) -> Result<(KnotHost, RepoSlug), XrpcError> { 2660 2660 + ) -> Result<RepoTarget, XrpcError> { 2644 2661 let rkey: Option<Rkey<DefaultStr>> = repo_uri.rkey().map(|r| r.clone().into_static()); 2645 2662 let (body, did) = resolve(state, ExpectedNsid::from_static(RepoRecord::NSID), repo_uri).await?; 2646 2663 let value: Repo<DefaultStr> = serde_json::from_slice(&body.value) ··· 2652 2669 })?; 2653 2670 let slug = RepoSlug::new(&did, &name) 2654 2671 .map_err(|e| XrpcError::InvalidRecord(format!("repo slug: {e}")))?; 2655 2655 - Ok((host, slug)) 2672 2672 + Ok(RepoTarget { 2673 2673 + host, 2674 2674 + slug, 2675 2675 + repo_did: value.repo_did, 2676 2676 + }) 2656 2677 } 2657 2678 2658 2679 fn pick_human_slug(rkey: Option<&Rkey<DefaultStr>>, name: Option<&str>) -> Option<String> { ··· 2670 2691 socket: SocketPeer, 2671 2692 address: &ClientAddress, 2672 2693 ) -> HeaderMap { 2673 2673 - let forwarded = FORWARDED_REQUEST_HEADERS 2694 2694 + let forwarded = RANGE_OR_CONDITIONAL_HEADERS 2674 2695 .iter() 2675 2696 .fold(HeaderMap::new(), |mut acc, name| { 2676 2697 if let Some(value) = client.get(*name) { ··· 2704 2725 response 2705 2726 } 2706 2727 2707 2707 - async fn dispatch_proxy( 2708 2708 - state: AppState, 2728 2728 + async fn dispatch_knot( 2729 2729 + state: &AppState, 2730 2730 + nsid: &Nsid<DefaultStr>, 2731 2731 + host: &KnotHost, 2732 2732 + query: &[(&str, &str)], 2709 2733 headers: HeaderMap, 2710 2710 - socket: SocketPeer, 2711 2711 - nsid: Nsid<DefaultStr>, 2712 2712 - host: KnotHost, 2713 2713 - params: ProxyParams, 2714 2734 ) -> Result<Response, XrpcError> { 2715 2715 - let forward: Vec<(&str, &str)> = params 2716 2716 - .iter() 2717 2717 - .map(|(k, v)| (k.as_str(), v.as_str())) 2718 2718 - .collect(); 2719 2719 - let allowed = filter_request_headers(&headers, socket, &state.client_address); 2720 2735 let upstream = state 2721 2736 .knots 2722 2722 - .forward(&host, &nsid, &forward, allowed) 2737 2737 + .forward(host, nsid, query, headers) 2723 2738 .await 2724 2739 .map_err(map_proxy_error)?; 2725 2740 Ok(upstream_to_axum(upstream)) 2726 2741 } 2727 2742 2743 2743 + async fn dispatch_mirror( 2744 2744 + state: &AppState, 2745 2745 + nsid: &Nsid<DefaultStr>, 2746 2746 + target: &RepoTarget, 2747 2747 + query: &[(&str, &str)], 2748 2748 + headers: &HeaderMap, 2749 2749 + ) -> Option<Response> { 2750 2750 + let mirror = state.mirror.as_ref()?; 2751 2751 + let repo_did = target.repo_did.as_ref()?; 2752 2752 + if RANGE_OR_CONDITIONAL_HEADERS 2753 2753 + .iter() 2754 2754 + .any(|name| headers.contains_key(*name)) 2755 2755 + { 2756 2756 + return None; 2757 2757 + } 2758 2758 + let mirror_nsid = MirrorNsid::route(nsid.as_ref(), query)?; 2759 2759 + let refused = match mirror 2760 2760 + .forward(&mirror_nsid, repo_did, query, headers.clone()) 2761 2761 + .await 2762 2762 + { 2763 2763 + Ok(upstream) if !upstream.status().is_client_error() => { 2764 2764 + return Some(upstream_to_axum(upstream)); 2765 2765 + } 2766 2766 + Ok(upstream) => { 2767 2767 + let status = upstream.status(); 2768 2768 + upstream.discard().await; 2769 2769 + status.to_string() 2770 2770 + } 2771 2771 + Err(KnotProxyError::Upstream(status)) => status.to_string(), 2772 2772 + Err(err @ KnotProxyError::CircuitOpen) => err.to_string(), 2773 2773 + Err(err) => { 2774 2774 + tracing::warn!( 2775 2775 + nsid = mirror_nsid.as_str(), 2776 2776 + repo = repo_did.as_ref(), 2777 2777 + error = %err, 2778 2778 + "mirror call failed, asking the knot", 2779 2779 + ); 2780 2780 + return None; 2781 2781 + } 2782 2782 + }; 2783 2783 + tracing::debug!( 2784 2784 + nsid = mirror_nsid.as_str(), 2785 2785 + repo = repo_did.as_ref(), 2786 2786 + refused, 2787 2787 + "mirror couldn't serve this repo, asking the knot", 2788 2788 + ); 2789 2789 + None 2790 2790 + } 2791 2791 + 2728 2792 fn extract_param( 2729 2793 params: ProxyParams, 2730 2794 key: &str, ··· 2751 2815 let (repo_raw, rest) = extract_param(params, REPO_PARAM)? 2752 2816 .ok_or_else(|| XrpcError::InvalidParams("missing repo".into()))?; 2753 2817 let repo_uri = parse_uri(&repo_raw)?; 2754 2754 - let (host, slug) = resolve_knot_target(&state, repo_uri).await?; 2755 2755 - let forward = rest 2818 2818 + let target = resolve_knot_target(&state, repo_uri).await?; 2819 2819 + let allowed = filter_request_headers(&headers, socket, &state.client_address); 2820 2820 + let query: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); 2821 2821 + if let Some(response) = dispatch_mirror(&state, &nsid, &target, &query, &allowed).await { 2822 2822 + return Ok(response); 2823 2823 + } 2824 2824 + let forward: Vec<(&str, &str)> = query 2756 2825 .into_iter() 2757 2757 - .chain(std::iter::once(( 2758 2758 - REPO_PARAM.to_owned(), 2759 2759 - slug.as_str().to_owned(), 2760 2760 - ))) 2826 2826 + .chain(std::iter::once((REPO_PARAM, target.slug.as_str()))) 2761 2827 .collect(); 2762 2762 - dispatch_proxy(state, headers, socket, nsid, host, forward).await 2828 2828 + dispatch_knot(&state, &nsid, &target.host, &forward, allowed).await 2763 2829 } 2764 2830 2765 2831 async fn proxy_knot_handler( ··· 2769 2835 params: ProxyParams, 2770 2836 nsid: Nsid<DefaultStr>, 2771 2837 ) -> Result<Response, XrpcError> { 2772 2772 - let (knot_raw, forward) = extract_param(params, KNOT_HOST_PARAM)? 2838 2838 + let (knot_raw, rest) = extract_param(params, KNOT_HOST_PARAM)? 2773 2839 .ok_or_else(|| XrpcError::InvalidParams("missing knot".into()))?; 2774 2840 let host = 2775 2841 KnotHost::parse(&knot_raw).map_err(|e| XrpcError::InvalidParams(format!("knot: {e}")))?; 2776 2842 validate_client_supplied_knot(&state, &host)?; 2777 2777 - dispatch_proxy(state, headers, socket, nsid, host, forward).await 2843 2843 + let allowed = filter_request_headers(&headers, socket, &state.client_address); 2844 2844 + let forward: Vec<(&str, &str)> = rest.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect(); 2845 2845 + dispatch_knot(&state, &nsid, &host, &forward, allowed).await 2778 2846 }

+358

bobbin/crates/xrpc/tests/mirror_proxy.rs

View file Reviewed

··· 1 1 + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; 2 2 + use std::sync::Arc; 3 3 + use std::time::Duration; 4 4 + 5 5 + use axum::body::{Body, to_bytes}; 6 6 + use axum::extract::ConnectInfo; 7 7 + use bobbin_edge_index::{CoverageWatch, EdgeStore, StateIndex}; 8 8 + use bobbin_knot_proxy::{KnotHttpConfig, KnotProxy, KnotProxyConfig, MirrorProxy}; 9 9 + use bobbin_record_lru::{CacheCapacity, LruRecordStore}; 10 10 + use bobbin_resolver::RepoIdResolver; 11 11 + use bobbin_runtime::{RuntimeHasher, SystemClock}; 12 12 + use bobbin_search::{DEFAULT_WRITER_HEAP_BYTES, SearchIndex, SearchReader}; 13 13 + use bobbin_slingshot_client::SlingshotClient; 14 14 + use bobbin_xrpc::{AppState, router}; 15 15 + use http::{Request, StatusCode}; 16 16 + use serde_json::json; 17 17 + use tower::ServiceExt; 18 18 + use url::Url; 19 19 + use url::form_urlencoded::byte_serialize; 20 20 + use wiremock::matchers::{method, path, query_param}; 21 21 + use wiremock::{Mock, MockServer, ResponseTemplate}; 22 22 + 23 23 + const CID: &str = "bafyreieqygohnz2zqyvtvktbjpvhutphobcmbsnt4q5lc36ri7vpcmoz4i"; 24 24 + const SOCKET: SocketAddr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 4321); 25 25 + const OWNER: &str = "did:plc:nel"; 26 26 + const RKEY: &str = "periwinkle"; 27 27 + const REPO_DID: &str = "did:plc:periwinkle"; 28 28 + const REPO_URI: &str = "at://did:plc:nel/sh.tangled.repo/periwinkle"; 29 29 + 30 30 + const FROM_MIRROR: &str = r#"{"served_by":"mirror"}"#; 31 31 + const FROM_KNOT: &str = r#"{"served_by":"knot"}"#; 32 32 + 33 33 + enum Mirror { 34 34 + Off, 35 35 + Live, 36 36 + Unreachable, 37 37 + } 38 38 + 39 39 + fn closed_port() -> String { 40 40 + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); 41 41 + let addr = listener.local_addr().unwrap(); 42 42 + drop(listener); 43 43 + format!("http://{addr}") 44 44 + } 45 45 + 46 46 + fn enc(s: &str) -> String { 47 47 + byte_serialize(s.as_bytes()).collect() 48 48 + } 49 49 + 50 50 + fn ok_from_mirror() -> ResponseTemplate { 51 51 + ResponseTemplate::new(200).set_body_raw(FROM_MIRROR, "application/json") 52 52 + } 53 53 + 54 54 + async fn paths(server: &MockServer) -> Vec<String> { 55 55 + server 56 56 + .received_requests() 57 57 + .await 58 58 + .unwrap() 59 59 + .iter() 60 60 + .map(|r| r.url.path().to_owned()) 61 61 + .collect() 62 62 + } 63 63 + 64 64 + struct Harness { 65 65 + _slingshot: MockServer, 66 66 + knot: MockServer, 67 67 + mirror: MockServer, 68 68 + state: AppState, 69 69 + } 70 70 + 71 71 + impl Harness { 72 72 + async fn new(setting: Mirror, repo_did: Option<&str>) -> Self { 73 73 + let slingshot = MockServer::start().await; 74 74 + let knot = MockServer::start().await; 75 75 + let mirror = MockServer::start().await; 76 76 + let clock = Arc::new(SystemClock::new()); 77 77 + let mirror_proxy = match setting { 78 78 + Mirror::Off => None, 79 79 + Mirror::Live => Some(mirror.uri()), 80 80 + Mirror::Unreachable => Some(closed_port()), 81 81 + } 82 82 + .map(|url| { 83 83 + Arc::new( 84 84 + MirrorProxy::new( 85 85 + &Url::parse(&url).unwrap(), 86 86 + clock.clone(), 87 87 + RuntimeHasher::default(), 88 88 + ) 89 89 + .unwrap(), 90 90 + ) 91 91 + }); 92 92 + let state = AppState::new( 93 93 + Arc::new(LruRecordStore::new(CacheCapacity::from_bytes(64 * 1024))), 94 94 + SlingshotClient::with_default_http(Url::parse(&slingshot.uri()).unwrap()).unwrap(), 95 95 + Arc::new(EdgeStore::new(RuntimeHasher::default())), 96 96 + Arc::new(StateIndex::new(RuntimeHasher::default())), 97 97 + Arc::new(StateIndex::new(RuntimeHasher::default())), 98 98 + Arc::new(CoverageWatch::new()), 99 99 + Arc::new( 100 100 + KnotProxy::new( 101 101 + KnotProxyConfig { 102 102 + allow_private_hosts: true, 103 103 + require_https: false, 104 104 + ..KnotProxyConfig::default() 105 105 + }, 106 106 + KnotHttpConfig { 107 107 + connect_timeout: Duration::from_millis(500), 108 108 + read_timeout: Duration::from_secs(2), 109 109 + }, 110 110 + clock.clone(), 111 111 + RuntimeHasher::default(), 112 112 + ) 113 113 + .unwrap(), 114 114 + ), 115 115 + Arc::new(SearchIndex::new(DEFAULT_WRITER_HEAP_BYTES, clock).unwrap()) 116 116 + as Arc<dyn SearchReader>, 117 117 + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), 118 118 + ) 119 119 + .with_mirror(mirror_proxy); 120 120 + 121 121 + let mut record = json!({ 122 122 + "$type": "sh.tangled.repo", 123 123 + "createdAt": "2026-05-01T00:00:00Z", 124 124 + "knot": knot.uri(), 125 125 + "name": "periwinkle", 126 126 + }); 127 127 + if let Some(d) = repo_did { 128 128 + record["repoDid"] = json!(d); 129 129 + } 130 130 + Mock::given(method("GET")) 131 131 + .and(path("/xrpc/com.atproto.repo.getRecord")) 132 132 + .and(query_param("repo", OWNER)) 133 133 + .and(query_param("collection", "sh.tangled.repo")) 134 134 + .and(query_param("rkey", RKEY)) 135 135 + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ 136 136 + "uri": REPO_URI, 137 137 + "cid": CID, 138 138 + "value": record, 139 139 + }))) 140 140 + .mount(&slingshot) 141 141 + .await; 142 142 + 143 143 + Self { 144 144 + _slingshot: slingshot, 145 145 + knot, 146 146 + mirror, 147 147 + state, 148 148 + } 149 149 + } 150 150 + 151 151 + async fn with_mirror() -> Self { 152 152 + Self::new(Mirror::Live, Some(REPO_DID)).await 153 153 + } 154 154 + 155 155 + async fn mount_mirror(&self, nsid: &str, response: ResponseTemplate) { 156 156 + Mock::given(method("GET")) 157 157 + .and(path(format!("/xrpc/{nsid}"))) 158 158 + .respond_with(response) 159 159 + .mount(&self.mirror) 160 160 + .await; 161 161 + } 162 162 + 163 163 + async fn mount_knot(&self, nsid: &str) { 164 164 + Mock::given(method("GET")) 165 165 + .and(path(format!("/xrpc/{nsid}"))) 166 166 + .respond_with(ResponseTemplate::new(200).set_body_raw(FROM_KNOT, "application/json")) 167 167 + .mount(&self.knot) 168 168 + .await; 169 169 + } 170 170 + 171 171 + async fn served_by(&self, nsid: &str, query: &str) -> String { 172 172 + self.served_by_with(nsid, query, &[]).await 173 173 + } 174 174 + 175 175 + async fn served_by_with(&self, nsid: &str, query: &str, headers: &[(&str, &str)]) -> String { 176 176 + let target = format!("/xrpc/{nsid}?repo={}&{query}", enc(REPO_URI)); 177 177 + let request = headers.iter().fold( 178 178 + Request::builder() 179 179 + .uri(target) 180 180 + .extension(ConnectInfo(SOCKET)), 181 181 + |builder, (name, value)| builder.header(*name, *value), 182 182 + ); 183 183 + let resp = router(self.state.clone()) 184 184 + .oneshot(request.body(Body::empty()).unwrap()) 185 185 + .await 186 186 + .expect("router infallible"); 187 187 + assert_eq!(resp.status(), StatusCode::OK, "{nsid}?{query}"); 188 188 + let body = to_bytes(resp.into_body(), 64 * 1024).await.unwrap(); 189 189 + String::from_utf8(body.to_vec()).unwrap() 190 190 + } 191 191 + } 192 192 + 193 193 + #[tokio::test] 194 194 + async fn tree_reads_the_mirror_keyed_on_the_repo_did() { 195 195 + let h = Harness::with_mirror().await; 196 196 + Mock::given(method("GET")) 197 197 + .and(path("/xrpc/sh.tangled.git.temp.getTree")) 198 198 + .and(query_param("repo", REPO_DID)) 199 199 + .and(query_param("ref", "main")) 200 200 + .respond_with(ok_from_mirror()) 201 201 + .mount(&h.mirror) 202 202 + .await; 203 203 + h.mount_knot("sh.tangled.repo.tree").await; 204 204 + 205 205 + assert_eq!( 206 206 + h.served_by("sh.tangled.repo.tree", "ref=main").await, 207 207 + FROM_MIRROR, 208 208 + "the mirror answers a read keyed on the repo did", 209 209 + ); 210 210 + assert!( 211 211 + paths(&h.knot).await.is_empty(), 212 212 + "the knot must stay untouched" 213 213 + ); 214 214 + } 215 215 + 216 216 + #[tokio::test] 217 217 + async fn every_shape_compatible_request_reads_the_mirror() { 218 218 + #[rustfmt::skip] 219 219 + let routed = [ 220 220 + ("sh.tangled.repo.branches", "sh.tangled.git.temp.listBranches", ""), 221 221 + ("sh.tangled.repo.log", "sh.tangled.git.temp.listCommits", "ref=main"), 222 222 + ("sh.tangled.repo.log", "sh.tangled.git.temp.listCommits", "ref=main&path="), 223 223 + ("sh.tangled.repo.tag", "sh.tangled.git.temp.getTag", "tag=v1"), 224 224 + ("sh.tangled.repo.tags", "sh.tangled.git.temp.listTags", ""), 225 225 + ("sh.tangled.repo.tree", "sh.tangled.git.temp.getTree", "ref=main"), 226 226 + ("sh.tangled.repo.tree", "sh.tangled.git.temp.getTree", "ref=main&path=crates"), 227 227 + ]; 228 228 + for (knot_nsid, mirror_nsid, query) in routed { 229 229 + let h = Harness::with_mirror().await; 230 230 + h.mount_mirror(mirror_nsid, ok_from_mirror()).await; 231 231 + h.mount_knot(knot_nsid).await; 232 232 + assert_eq!( 233 233 + h.served_by(knot_nsid, query).await, 234 234 + FROM_MIRROR, 235 235 + "{knot_nsid}?{query} must read the mirror", 236 236 + ); 237 237 + assert_eq!(paths(&h.mirror).await, vec![format!("/xrpc/{mirror_nsid}")]); 238 238 + } 239 239 + } 240 240 + 241 241 + #[tokio::test] 242 242 + async fn every_request_the_mirror_answers_in_another_shape_reads_the_knot() { 243 243 + #[rustfmt::skip] 244 244 + let refused = [ 245 245 + ("sh.tangled.repo.blob", "ref=main&path=x", "the mirror serves content types the knot answers with 403"), 246 246 + ("sh.tangled.repo.blob", "ref=main&path=x&raw=true", "raw doesn't exempt the blob"), 247 247 + ("sh.tangled.repo.archive", "ref=main", "a resume would splice knot bytes onto a mirror tarball"), 248 248 + ("sh.tangled.repo.log", "ref=main&path=crates/xrpc", "the mirror ignores path"), 249 249 + ("sh.tangled.repo.branch", "name=main", "the mirror answers branch in another shape"), 250 250 + ("sh.tangled.repo.languages", "ref=main", "the mirror answers languages in another shape"), 251 251 + ("sh.tangled.repo.compare", "", "outside the routing table"), 252 252 + ("sh.tangled.repo.describeRepo", "", "outside the routing table"), 253 253 + ("sh.tangled.repo.diff", "", "outside the routing table"), 254 254 + ("sh.tangled.repo.getDefaultBranch", "", "outside the routing table"), 255 255 + ("sh.tangled.repo.listSecrets", "", "outside the routing table"), 256 256 + ]; 257 257 + for (nsid, query, why) in refused { 258 258 + let h = Harness::with_mirror().await; 259 259 + h.mount_knot(nsid).await; 260 260 + assert_eq!(h.served_by(nsid, query).await, FROM_KNOT, "{nsid}: {why}"); 261 261 + assert!(paths(&h.mirror).await.is_empty(), "{nsid}: {why}"); 262 262 + } 263 263 + } 264 264 + 265 265 + #[tokio::test] 266 266 + async fn every_mirror_refusal_reads_the_knot() { 267 267 + for status in [400, 403, 404, 503] { 268 268 + let h = Harness::with_mirror().await; 269 269 + h.mount_mirror( 270 270 + "sh.tangled.git.temp.listBranches", 271 271 + ResponseTemplate::new(status).set_body_json(json!({"error": "BadRequest"})), 272 272 + ) 273 273 + .await; 274 274 + h.mount_knot("sh.tangled.repo.branches").await; 275 275 + assert_eq!( 276 276 + h.served_by("sh.tangled.repo.branches", "limit=500").await, 277 277 + FROM_KNOT, 278 278 + "the knot must answer after a mirror {status}", 279 279 + ); 280 280 + } 281 281 + 282 282 + let h = Harness::new(Mirror::Unreachable, Some(REPO_DID)).await; 283 283 + h.mount_knot("sh.tangled.repo.branches").await; 284 284 + assert_eq!( 285 285 + h.served_by("sh.tangled.repo.branches", "").await, 286 286 + FROM_KNOT, 287 287 + "the knot must answer when the mirror is unreachable", 288 288 + ); 289 289 + } 290 290 + 291 291 + #[tokio::test] 292 292 + async fn a_ranged_or_conditional_request_skips_the_mirror() { 293 293 + for (header, value) in [ 294 294 + ("range", "bytes=0-99"), 295 295 + ("if-range", "bytes=0-99"), 296 296 + ("if-none-match", "\"cafe\""), 297 297 + ("if-modified-since", "Wed, 01 Jul 2026 00:00:00 GMT"), 298 298 + ] { 299 299 + let h = Harness::with_mirror().await; 300 300 + h.mount_mirror("sh.tangled.git.temp.getTree", ok_from_mirror()) 301 301 + .await; 302 302 + h.mount_knot("sh.tangled.repo.tree").await; 303 303 + 304 304 + assert_eq!( 305 305 + h.served_by_with("sh.tangled.repo.tree", "ref=main", &[(header, value)]) 306 306 + .await, 307 307 + FROM_KNOT, 308 308 + "only the knot can answer a {header} it issued", 309 309 + ); 310 310 + assert!(paths(&h.mirror).await.is_empty(), "{header}"); 311 311 + } 312 312 + } 313 313 + 314 314 + #[tokio::test] 315 315 + async fn a_request_reads_the_knot_when_bobbin_wont_ask_the_mirror() { 316 316 + #[rustfmt::skip] 317 317 + let unasked = [ 318 318 + (Mirror::Live, None, "the mirror keys on a repoDid this record doesn't have"), 319 319 + (Mirror::Off, Some(REPO_DID), "an unset mirror.url leaves every call on the knot"), 320 320 + ]; 321 321 + for (setting, repo_did, why) in unasked { 322 322 + let h = Harness::new(setting, repo_did).await; 323 323 + h.mount_mirror("sh.tangled.git.temp.getTree", ok_from_mirror()) 324 324 + .await; 325 325 + h.mount_knot("sh.tangled.repo.tree").await; 326 326 + 327 327 + assert_eq!( 328 328 + h.served_by("sh.tangled.repo.tree", "ref=main").await, 329 329 + FROM_KNOT, 330 330 + "{why}", 331 331 + ); 332 332 + assert!(paths(&h.mirror).await.is_empty(), "{why}"); 333 333 + } 334 334 + } 335 335 + 336 336 + #[tokio::test] 337 337 + async fn the_knot_keyed_endpoints_never_read_the_mirror() { 338 338 + let h = Harness::with_mirror().await; 339 339 + Mock::given(method("GET")) 340 340 + .and(path("/xrpc/sh.tangled.knot.version")) 341 341 + .respond_with(ResponseTemplate::new(200).set_body_raw(FROM_KNOT, "application/json")) 342 342 + .mount(&h.knot) 343 343 + .await; 344 344 + 345 345 + let target = format!("/xrpc/sh.tangled.knot.version?knot={}", enc(&h.knot.uri())); 346 346 + let resp = router(h.state.clone()) 347 347 + .oneshot( 348 348 + Request::builder() 349 349 + .uri(target) 350 350 + .extension(ConnectInfo(SOCKET)) 351 351 + .body(Body::empty()) 352 352 + .unwrap(), 353 353 + ) 354 354 + .await 355 355 + .unwrap(); 356 356 + assert_eq!(resp.status(), StatusCode::OK); 357 357 + assert!(paths(&h.mirror).await.is_empty()); 358 358 + }

+4

bobbin/example.toml

View file Reviewed

··· 162 162 # Default value: true 163 163 #require_https = true 164 164 165 165 + [mirror] 166 166 + # Can also be specified via environment variable `BOBBIN_MIRROR_URL`. 167 167 + #url = 168 168 + 165 169 [log] 166 170 # Log emitter format. `text` produces human-readable output for local 167 171 # development. `json` emits one structured object per line for log

+15

knotmirror/xrpc/proxy.go

View file Reviewed

··· 7 7 "fmt" 8 8 "io" 9 9 "maps" 10 10 + "net" 10 11 "net/http" 11 12 "net/url" 12 13 "path" ··· 35 36 tangled.GitTempListLanguagesNSID: tangled.RepoLanguagesNSID, 36 37 tangled.GitTempGetBlobNSID: tangled.RepoBlobNSID, 37 38 } 39 39 + 40 40 + const forwardedForHeader = "X-Forwarded-For" 38 41 39 42 var hopByHopHeaders = map[string]bool{ 40 43 "Connection": true, ··· 132 135 x.logger.Warn("proxy: failed to build request", "target", target, "err", err) 133 136 return false 134 137 } 138 138 + req.Header.Set(forwardedForHeader, forwardedFor(r)) 135 139 136 140 resp, err := x.httpClient.Do(req) 137 141 if err != nil { ··· 155 159 156 160 x.logger.Info("proxy: served from knot", "repo", repoDid, "knot", knot.baseURL, "status", resp.StatusCode) 157 161 return true 162 162 + } 163 163 + 164 164 + func forwardedFor(r *http.Request) string { 165 165 + peer := r.RemoteAddr 166 166 + if host, _, err := net.SplitHostPort(r.RemoteAddr); err == nil { 167 167 + peer = host 168 168 + } 169 169 + chain := lo.Filter(r.Header.Values(forwardedForHeader), func(entry string, _ int) bool { 170 170 + return strings.TrimSpace(entry) != "" 171 171 + }) 172 172 + return strings.Join(append(chain, peer), ", ") 158 173 } 159 174 160 175 func (x *Xrpc) forwardSuspended(next http.Handler) http.Handler {

+38

knotmirror/xrpc/proxy_test.go

View file Reviewed

··· 1 1 + package xrpc 2 2 + 3 3 + import ( 4 4 + "net/http/httptest" 5 5 + "testing" 6 6 + 7 7 + "github.com/stretchr/testify/assert" 8 8 + ) 9 9 + 10 10 + func TestForwardedForAppendsThePeerToTheChain(t *testing.T) { 11 11 + cases := []struct { 12 12 + name string 13 13 + remote string 14 14 + chain []string 15 15 + want string 16 16 + }{ 17 17 + {"a direct caller is the whole chain", "203.0.113.7:52344", nil, "203.0.113.7"}, 18 18 + {"a portless remote address passes through", "203.0.113.7", nil, "203.0.113.7"}, 19 19 + {"bobbin's client address stays left of bobbin", "198.51.100.4:41000", []string{"203.0.113.7"}, "203.0.113.7, 198.51.100.4"}, 20 20 + {"a chain split across header lines joins into a single value", "198.51.100.4:41000", []string{"203.0.113.7", "192.0.2.9"}, "203.0.113.7, 192.0.2.9, 198.51.100.4"}, 21 21 + {"a blank entry never leaves a gap the knot has to skip", "198.51.100.4:41000", []string{"", " ", "203.0.113.7"}, "203.0.113.7, 198.51.100.4"}, 22 22 + {"an ipv6 peer loses its port and keeps its colons", "[2001:db8::5]:41000", []string{"203.0.113.7"}, "203.0.113.7, 2001:db8::5"}, 23 23 + {"a forged entry stays left of the address that sent it", "203.0.113.7:52344", []string{"192.0.2.9"}, "192.0.2.9, 203.0.113.7"}, 24 24 + } 25 25 + 26 26 + for _, c := range cases { 27 27 + t.Run(c.name, func(t *testing.T) { 28 28 + r := httptest.NewRequest("GET", "/xrpc/sh.tangled.git.temp.getTree?repo=did:plc:limpet", nil) 29 29 + r.RemoteAddr = c.remote 30 30 + for _, entry := range c.chain { 31 31 + r.Header.Add(forwardedForHeader, entry) 32 32 + } 33 33 + 34 34 + assert.Equal(t, c.want, forwardedFor(r)) 35 35 + assert.Equal(t, c.chain, r.Header.Values(forwardedForHeader), "the caller's own header must survive the read") 36 36 + }) 37 37 + } 38 38 + }