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