alice.pds.demo.boltless.dev
This repository has no description
did:plc:3fwecdnvtcscjnrx2p4n7a…
1use std::pin::Pin;
2use std::sync::Arc;
3use std::task::{Context, Poll};
4use std::time::Duration;
5
6use bobbin_runtime::{
7 BodyStream as InnerBodyStream, Clock, HttpRequest, HttpResponseHead, HttpTransport,
8 NetworkError, ReqwestHttp, RuntimeHasher,
9};
10use bytes::Bytes;
11use futures::{Stream, StreamExt};
12use http::{HeaderMap, StatusCode};
13use jacquard_common::BosStr;
14use jacquard_common::types::nsid::Nsid;
15use reqwest::{Client, redirect::Policy};
16use scc::HashMap as SccMap;
17use thiserror::Error;
18use url::Url;
19
20mod breaker;
21mod dns;
22mod host;
23mod mirror;
24
25pub use breaker::{Breaker, BreakerPermit, CircuitOpen, FailureThreshold, ThresholdError};
26pub use dns::PrivateAddressFilter;
27pub use host::{KnotHost, KnotHostError, PrivateHostReason, RepoSlug, RepoSlugError, classify_ip};
28pub use mirror::{MirrorNsid, MirrorProxy, MirrorProxyError};
29
30const USER_AGENT: &str = concat!("bobbin/", env!("CARGO_PKG_VERSION"));
31const HTTPS_SCHEME: &str = "https";
32const DISCARD_BUDGET_BYTES: usize = 64 * 1024;
33
34#[derive(Clone, Debug)]
35pub struct KnotProxyConfig {
36 pub failure_threshold: FailureThreshold,
37 pub cooldown: Duration,
38 pub allow_private_hosts: bool,
39 pub require_https: bool,
40}
41
42impl Default for KnotProxyConfig {
43 fn default() -> Self {
44 Self {
45 failure_threshold: FailureThreshold::new(5).expect("nonzero literal"),
46 cooldown: Duration::from_secs(30),
47 allow_private_hosts: false,
48 require_https: true,
49 }
50 }
51}
52
53#[derive(Clone, Copy, Debug)]
54pub struct KnotHttpConfig {
55 pub connect_timeout: Duration,
56 pub read_timeout: Duration,
57}
58
59impl Default for KnotHttpConfig {
60 fn default() -> Self {
61 Self {
62 connect_timeout: Duration::from_secs(5),
63 read_timeout: Duration::from_secs(60),
64 }
65 }
66}
67
68#[derive(Debug, Error)]
69pub enum KnotProxyError {
70 #[error("circuit breaker open")]
71 CircuitOpen,
72 #[error("blocked: host {host} resolves to {reason} address space")]
73 BlockedHost {
74 host: String,
75 reason: PrivateHostReason,
76 },
77 #[error("blocked: knot {host} requires https, got plaintext http")]
78 PlaintextHttp { host: String },
79 #[error("connect failed: {0}")]
80 Connect(String),
81 #[error("upstream read timed out: {0}")]
82 Timeout(String),
83 #[error("redirect refused: {0}")]
84 Redirect(String),
85 #[error("transport: {0}")]
86 Transport(String),
87 #[error("upstream returned status {0}")]
88 Upstream(StatusCode),
89}
90
91pub struct KnotProxy {
92 http: Arc<dyn HttpTransport>,
93 breakers: SccMap<KnotHost, Arc<Breaker>, RuntimeHasher>,
94 threshold: FailureThreshold,
95 cooldown: Duration,
96 allow_private_hosts: bool,
97 require_https: bool,
98 clock: Arc<dyn Clock>,
99}
100
101impl KnotProxy {
102 pub fn new(
103 config: KnotProxyConfig,
104 http: KnotHttpConfig,
105 clock: Arc<dyn Clock>,
106 hasher: RuntimeHasher,
107 ) -> Result<Self, reqwest::Error> {
108 let resolver = Arc::new(dns::PrivateAddressFilter::new(config.allow_private_hosts));
109 let client = Client::builder()
110 .user_agent(USER_AGENT)
111 .connect_timeout(http.connect_timeout)
112 .read_timeout(http.read_timeout)
113 .redirect(Policy::none())
114 .no_gzip()
115 .no_brotli()
116 .no_deflate()
117 .dns_resolver(resolver)
118 .build()?;
119 Ok(Self::with_transport(
120 ReqwestHttp::shared(client),
121 config,
122 clock,
123 hasher,
124 ))
125 }
126
127 pub fn with_transport(
128 http: Arc<dyn HttpTransport>,
129 config: KnotProxyConfig,
130 clock: Arc<dyn Clock>,
131 hasher: RuntimeHasher,
132 ) -> Self {
133 Self {
134 http,
135 breakers: SccMap::with_hasher(hasher),
136 threshold: config.failure_threshold,
137 cooldown: config.cooldown,
138 allow_private_hosts: config.allow_private_hosts,
139 require_https: config.require_https,
140 clock,
141 }
142 }
143
144 pub fn allows_private_hosts(&self) -> bool {
145 self.allow_private_hosts
146 }
147
148 pub fn requires_https(&self) -> bool {
149 self.require_https
150 }
151
152 pub async fn forward<S: BosStr + AsRef<str>>(
153 &self,
154 host: &KnotHost,
155 nsid: &Nsid<S>,
156 query: &[(&str, &str)],
157 headers: HeaderMap,
158 ) -> Result<ProxyResponse, KnotProxyError> {
159 self.guard_host(host)?;
160 let breaker = self.breaker_for(host).await;
161 let permit = breaker
162 .try_acquire()
163 .map_err(|_: CircuitOpen| KnotProxyError::CircuitOpen)?;
164 let url = build_xrpc_url(host, nsid, query);
165 let outcome = self.http.execute(HttpRequest { url, headers }).await;
166 classify(outcome, permit)
167 }
168
169 fn guard_host(&self, host: &KnotHost) -> Result<(), KnotProxyError> {
170 let host_str = || host.url().host_str().unwrap_or_default().to_owned();
171 if self.require_https && host.url().scheme() != HTTPS_SCHEME {
172 return Err(KnotProxyError::PlaintextHttp { host: host_str() });
173 }
174 if self.allow_private_hosts {
175 return Ok(());
176 }
177 match host.private_literal_reason() {
178 None => Ok(()),
179 Some(reason) => Err(KnotProxyError::BlockedHost {
180 host: host_str(),
181 reason,
182 }),
183 }
184 }
185
186 async fn breaker_for(&self, host: &KnotHost) -> Arc<Breaker> {
187 if let Some(existing) = self.breakers.read_async(host, |_, v| Arc::clone(v)).await {
188 return existing;
189 }
190 let entry = self.breakers.entry_async(host.clone()).await;
191 Arc::clone(
192 entry
193 .or_insert_with(|| {
194 Arc::new(Breaker::new(
195 self.threshold,
196 self.cooldown,
197 self.clock.clone(),
198 ))
199 })
200 .get(),
201 )
202 }
203}
204
205pub struct ProxyResponse {
206 status: StatusCode,
207 headers: HeaderMap,
208 body: InnerBodyStream,
209 permit: BreakerPermit,
210}
211
212impl std::fmt::Debug for ProxyResponse {
213 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
214 f.debug_struct("ProxyResponse")
215 .field("status", &self.status)
216 .field("headers", &self.headers)
217 .finish_non_exhaustive()
218 }
219}
220
221impl ProxyResponse {
222 pub fn status(&self) -> StatusCode {
223 self.status
224 }
225
226 pub fn headers(&self) -> &HeaderMap {
227 &self.headers
228 }
229
230 pub fn into_body_stream(self) -> BodyStream {
231 BodyStream::new(self.body, self.permit)
232 }
233
234 pub async fn discard(self) {
235 let mut stream = self.into_body_stream();
236 let mut seen = 0usize;
237 while seen < DISCARD_BUDGET_BYTES {
238 match stream.next().await {
239 None | Some(Err(_)) => return,
240 Some(Ok(chunk)) => seen = seen.saturating_add(chunk.len().max(1)),
241 }
242 }
243 }
244}
245
246pub struct BodyStream {
247 inner: InnerBodyStream,
248 permit: Option<BreakerPermit>,
249}
250
251impl BodyStream {
252 fn new(inner: InnerBodyStream, permit: BreakerPermit) -> Self {
253 Self {
254 inner,
255 permit: Some(permit),
256 }
257 }
258
259 fn resolve(&mut self, success: bool) {
260 if let Some(permit) = self.permit.take() {
261 if success {
262 permit.record_success();
263 } else {
264 permit.record_failure();
265 }
266 }
267 }
268}
269
270impl Stream for BodyStream {
271 type Item = Result<Bytes, NetworkError>;
272
273 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
274 let next = match self.inner.as_mut().poll_next(cx) {
275 Poll::Pending => return Poll::Pending,
276 Poll::Ready(item) => item,
277 };
278 match &next {
279 Some(Ok(_)) => {}
280 Some(Err(_)) => self.resolve(false),
281 None => self.resolve(true),
282 }
283 Poll::Ready(next)
284 }
285}
286
287fn build_xrpc_url<S: BosStr + AsRef<str>>(
288 host: &KnotHost,
289 nsid: &Nsid<S>,
290 query: &[(&str, &str)],
291) -> Url {
292 let mut url = host.xrpc_url(nsid);
293 {
294 let mut pairs = url.query_pairs_mut();
295 query.iter().for_each(|(k, v)| {
296 pairs.append_pair(k, v);
297 });
298 }
299 url
300}
301
302fn classify(
303 outcome: Result<HttpResponseHead, NetworkError>,
304 permit: BreakerPermit,
305) -> Result<ProxyResponse, KnotProxyError> {
306 match outcome {
307 Ok(head) if is_upstream_failure(head.status) => {
308 let status = head.status;
309 permit.record_failure();
310 Err(KnotProxyError::Upstream(status))
311 }
312 Ok(head) => Ok(ProxyResponse {
313 status: head.status,
314 headers: head.headers,
315 body: head.body,
316 permit,
317 }),
318 Err(err) => {
319 permit.record_failure();
320 Err(map_network(err))
321 }
322 }
323}
324
325fn is_upstream_failure(status: StatusCode) -> bool {
326 status.is_server_error() || is_unfollowable_redirect(status)
327}
328
329fn is_unfollowable_redirect(status: StatusCode) -> bool {
330 matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308)
331}
332
333fn map_network(err: NetworkError) -> KnotProxyError {
334 match err {
335 NetworkError::Timeout(msg) => KnotProxyError::Timeout(msg),
336 NetworkError::Connect(msg) => KnotProxyError::Connect(msg),
337 NetworkError::Redirect(msg) => KnotProxyError::Redirect(msg),
338 NetworkError::Transport(msg) | NetworkError::Body(msg) | NetworkError::Protocol(msg) => {
339 KnotProxyError::Transport(msg)
340 }
341 }
342}
343
344#[cfg(test)]
345mod tests {
346 use super::*;
347 use bobbin_runtime::SystemClock;
348 use futures::stream::TryStreamExt;
349 use jacquard_common::DefaultStr;
350 use tokio::io::AsyncWriteExt;
351 use wiremock::matchers::{method, path, query_param};
352 use wiremock::{Mock, MockServer, ResponseTemplate};
353
354 fn nsid(s: &'static str) -> Nsid<DefaultStr> {
355 Nsid::new_static(s).unwrap()
356 }
357
358 pub(crate) fn config_for_test() -> KnotProxyConfig {
359 KnotProxyConfig {
360 failure_threshold: FailureThreshold::new(2).unwrap(),
361 cooldown: Duration::from_millis(80),
362 allow_private_hosts: true,
363 require_https: false,
364 }
365 }
366
367 pub(crate) fn http_config_for_test() -> KnotHttpConfig {
368 KnotHttpConfig {
369 connect_timeout: Duration::from_millis(500),
370 read_timeout: Duration::from_secs(2),
371 }
372 }
373
374 fn proxy_for_test() -> KnotProxy {
375 KnotProxy::new(
376 config_for_test(),
377 http_config_for_test(),
378 Arc::new(SystemClock::new()),
379 RuntimeHasher::default(),
380 )
381 .unwrap()
382 }
383
384 async fn server() -> MockServer {
385 MockServer::start().await
386 }
387
388 fn host_of(server: &MockServer) -> KnotHost {
389 KnotHost::parse(&server.uri()).unwrap()
390 }
391
392 pub(crate) async fn drain(stream: BodyStream) -> Result<Bytes, NetworkError> {
393 let chunks: Vec<Bytes> = stream.try_collect().await?;
394 let total: usize = chunks.iter().map(|b| b.len()).sum();
395 let mut buf = bytes::BytesMut::with_capacity(total);
396 chunks.iter().for_each(|c| buf.extend_from_slice(c));
397 Ok(buf.freeze())
398 }
399
400 #[tokio::test]
401 async fn forwards_query_params_and_returns_body() {
402 let server = server().await;
403 Mock::given(method("GET"))
404 .and(path("/xrpc/sh.tangled.repo.blob"))
405 .and(query_param("repo", "did:plc:squid/barnacle"))
406 .and(query_param("ref", "main"))
407 .and(query_param("path", "README.md"))
408 .respond_with(
409 ResponseTemplate::new(200)
410 .insert_header("content-type", "application/json")
411 .set_body_string(r#"{"path":"README.md"}"#),
412 )
413 .mount(&server)
414 .await;
415
416 let proxy = proxy_for_test();
417 let resp = proxy
418 .forward(
419 &host_of(&server),
420 &nsid("sh.tangled.repo.blob"),
421 &[
422 ("repo", "did:plc:squid/barnacle"),
423 ("ref", "main"),
424 ("path", "README.md"),
425 ],
426 HeaderMap::new(),
427 )
428 .await
429 .expect("happy path");
430 assert_eq!(resp.status(), 200);
431 let body = drain(resp.into_body_stream()).await.unwrap();
432 assert_eq!(&body[..], br#"{"path":"README.md"}"#);
433 }
434
435 #[tokio::test]
436 async fn five_hundreds_open_breaker() {
437 let server = server().await;
438 Mock::given(method("GET"))
439 .and(path("/xrpc/sh.tangled.repo.blob"))
440 .respond_with(ResponseTemplate::new(503))
441 .mount(&server)
442 .await;
443
444 let proxy = proxy_for_test();
445 let host = host_of(&server);
446 let r1 = proxy
447 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
448 .await;
449 assert!(matches!(r1, Err(KnotProxyError::Upstream(_))));
450 let r2 = proxy
451 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
452 .await;
453 assert!(matches!(r2, Err(KnotProxyError::Upstream(_))));
454 let r3 = proxy
455 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
456 .await;
457 assert!(matches!(r3, Err(KnotProxyError::CircuitOpen)));
458 }
459
460 #[tokio::test]
461 async fn four_hundreds_do_not_open_breaker() {
462 let server = server().await;
463 Mock::given(method("GET"))
464 .and(path("/xrpc/sh.tangled.repo.blob"))
465 .respond_with(ResponseTemplate::new(404).set_body_string("not found"))
466 .mount(&server)
467 .await;
468
469 let proxy = proxy_for_test();
470 let host = host_of(&server);
471 let r1 = proxy
472 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
473 .await;
474 assert_eq!(r1.unwrap().status(), 404);
475 let r2 = proxy
476 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
477 .await;
478 assert_eq!(r2.unwrap().status(), 404);
479 let r3 = proxy
480 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
481 .await;
482 assert_eq!(
483 r3.unwrap().status(),
484 404,
485 "client errors must not trip breaker",
486 );
487 }
488
489 #[tokio::test]
490 async fn breaker_recovers_after_cooldown() {
491 let server = server().await;
492 Mock::given(method("GET"))
493 .and(path("/xrpc/sh.tangled.repo.blob"))
494 .respond_with(ResponseTemplate::new(503))
495 .up_to_n_times(2)
496 .mount(&server)
497 .await;
498 Mock::given(method("GET"))
499 .and(path("/xrpc/sh.tangled.repo.blob"))
500 .respond_with(
501 ResponseTemplate::new(200)
502 .insert_header("content-type", "application/json")
503 .set_body_string("ok"),
504 )
505 .mount(&server)
506 .await;
507
508 let proxy = proxy_for_test();
509 let host = host_of(&server);
510 let _ = proxy
511 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
512 .await;
513 let _ = proxy
514 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
515 .await;
516 assert!(matches!(
517 proxy
518 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
519 .await,
520 Err(KnotProxyError::CircuitOpen),
521 ));
522 tokio::time::sleep(Duration::from_millis(120)).await;
523 let recovered = proxy
524 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
525 .await
526 .expect("must recover after cooldown");
527 assert_eq!(recovered.status(), 200);
528 let body = drain(recovered.into_body_stream()).await.unwrap();
529 assert_eq!(&body[..], b"ok");
530 }
531
532 #[tokio::test]
533 async fn breakers_are_isolated_per_host() {
534 let bad = server().await;
535 let good = server().await;
536 Mock::given(method("GET"))
537 .and(path("/xrpc/sh.tangled.repo.blob"))
538 .respond_with(ResponseTemplate::new(503))
539 .mount(&bad)
540 .await;
541 Mock::given(method("GET"))
542 .and(path("/xrpc/sh.tangled.repo.blob"))
543 .respond_with(
544 ResponseTemplate::new(200)
545 .insert_header("content-type", "application/json")
546 .set_body_string("ok"),
547 )
548 .mount(&good)
549 .await;
550
551 let proxy = proxy_for_test();
552 let bad_host = host_of(&bad);
553 let good_host = host_of(&good);
554 let _ = proxy
555 .forward(
556 &bad_host,
557 &nsid("sh.tangled.repo.blob"),
558 &[],
559 HeaderMap::new(),
560 )
561 .await;
562 let _ = proxy
563 .forward(
564 &bad_host,
565 &nsid("sh.tangled.repo.blob"),
566 &[],
567 HeaderMap::new(),
568 )
569 .await;
570 assert!(matches!(
571 proxy
572 .forward(
573 &bad_host,
574 &nsid("sh.tangled.repo.blob"),
575 &[],
576 HeaderMap::new()
577 )
578 .await,
579 Err(KnotProxyError::CircuitOpen),
580 ));
581 let resp = proxy
582 .forward(
583 &good_host,
584 &nsid("sh.tangled.repo.blob"),
585 &[],
586 HeaderMap::new(),
587 )
588 .await
589 .expect("healthy host stays open");
590 assert_eq!(resp.status(), 200);
591 }
592
593 #[tokio::test]
594 async fn build_xrpc_url_appends_query() {
595 let host = KnotHost::parse("https://oyster.cafe").unwrap();
596 let url = build_xrpc_url(
597 &host,
598 &nsid("sh.tangled.repo.tree"),
599 &[("repo", "did:plc:squid/barnacle"), ("ref", "main")],
600 );
601 assert_eq!(
602 url.as_str(),
603 "https://oyster.cafe/xrpc/sh.tangled.repo.tree?repo=did%3Aplc%3Asquid%2Fbarnacle&ref=main",
604 );
605 }
606
607 #[tokio::test]
608 async fn rejects_private_host_by_default() {
609 let server = server().await;
610 let strict = KnotProxyConfig {
611 allow_private_hosts: false,
612 ..config_for_test()
613 };
614 let proxy = KnotProxy::new(
615 strict,
616 http_config_for_test(),
617 Arc::new(SystemClock::new()),
618 RuntimeHasher::default(),
619 )
620 .unwrap();
621 let err = proxy
622 .forward(
623 &host_of(&server),
624 &nsid("sh.tangled.repo.blob"),
625 &[],
626 HeaderMap::new(),
627 )
628 .await
629 .expect_err("loopback must be blocked under strict config");
630 assert!(matches!(err, KnotProxyError::BlockedHost { .. }));
631 }
632
633 #[tokio::test]
634 async fn rejects_plaintext_when_https_required() {
635 let server = server().await;
636 let strict = KnotProxyConfig {
637 require_https: true,
638 ..config_for_test()
639 };
640 let proxy = KnotProxy::new(
641 strict,
642 http_config_for_test(),
643 Arc::new(SystemClock::new()),
644 RuntimeHasher::default(),
645 )
646 .unwrap();
647 let err = proxy
648 .forward(
649 &host_of(&server),
650 &nsid("sh.tangled.repo.blob"),
651 &[],
652 HeaderMap::new(),
653 )
654 .await
655 .expect_err("plaintext http must be rejected under https-required");
656 assert!(
657 matches!(err, KnotProxyError::PlaintextHttp { .. }),
658 "got {err:?}",
659 );
660 }
661
662 #[tokio::test]
663 async fn transport_error_trips_breaker() {
664 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
665 let addr = listener.local_addr().unwrap();
666 drop(listener);
667 let dead = KnotHost::parse(&format!("http://{addr}")).unwrap();
668
669 let proxy = proxy_for_test();
670 let r1 = proxy
671 .forward(&dead, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
672 .await;
673 assert!(
674 r1.is_err(),
675 "transport must fail against closed port: {r1:?}"
676 );
677 let r2 = proxy
678 .forward(&dead, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
679 .await;
680 assert!(r2.is_err(), "second transport must fail: {r2:?}");
681 let r3 = proxy
682 .forward(&dead, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
683 .await;
684 assert!(
685 matches!(r3, Err(KnotProxyError::CircuitOpen)),
686 "transport failures must trip breaker, got {r3:?}",
687 );
688 }
689
690 #[tokio::test]
691 async fn redirects_surface_as_upstream_failure() {
692 let primary = server().await;
693 let secondary = server().await;
694 Mock::given(method("GET"))
695 .and(path("/xrpc/sh.tangled.repo.blob"))
696 .respond_with(
697 ResponseTemplate::new(302)
698 .insert_header("location", &format!("{}/secret", secondary.uri())),
699 )
700 .mount(&primary)
701 .await;
702 Mock::given(method("GET"))
703 .and(path("/secret"))
704 .respond_with(ResponseTemplate::new(200).set_body_string("leaked"))
705 .mount(&secondary)
706 .await;
707
708 let proxy = proxy_for_test();
709 let err = proxy
710 .forward(
711 &host_of(&primary),
712 &nsid("sh.tangled.repo.blob"),
713 &[],
714 HeaderMap::new(),
715 )
716 .await
717 .expect_err("302 must surface as upstream failure");
718 assert!(
719 matches!(err, KnotProxyError::Upstream(s) if s.as_u16() == 302),
720 "got {err:?}",
721 );
722 let received = secondary.received_requests().await.unwrap();
723 assert!(received.is_empty(), "secondary must never be dialled");
724 }
725
726 #[tokio::test]
727 async fn not_modified_passes_through() {
728 let server = server().await;
729 Mock::given(method("GET"))
730 .and(path("/xrpc/sh.tangled.repo.blob"))
731 .respond_with(ResponseTemplate::new(304).insert_header("etag", "\"v1\""))
732 .mount(&server)
733 .await;
734 let proxy = proxy_for_test();
735 let resp = proxy
736 .forward(
737 &host_of(&server),
738 &nsid("sh.tangled.repo.blob"),
739 &[],
740 HeaderMap::new(),
741 )
742 .await
743 .expect("304 is a cache validator, not a redirect");
744 assert_eq!(resp.status(), 304);
745 }
746
747 #[tokio::test]
748 async fn mid_stream_drop_records_breaker_failure() {
749 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
750 let addr = listener.local_addr().unwrap();
751 let server = tokio::spawn(async move {
752 async fn drop_after_partial(mut socket: tokio::net::TcpStream) {
753 let _ = socket
754 .write_all(
755 b"HTTP/1.1 200 OK\r\nContent-Length: 1024\r\nContent-Type: application/octet-stream\r\n\r\nabcd",
756 )
757 .await;
758 drop(socket);
759 }
760 let admit = || async {
761 let (socket, _) = listener.accept().await.ok()?;
762 drop_after_partial(socket).await;
763 Some(())
764 };
765 admit().await;
766 admit().await;
767 });
768
769 let host = KnotHost::parse(&format!("http://{addr}")).unwrap();
770 let proxy = proxy_for_test();
771
772 let r1 = proxy
773 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
774 .await
775 .expect("headers arrive even when body is truncated");
776 assert_eq!(r1.status(), 200);
777 let _ = drain(r1.into_body_stream()).await;
778
779 let r2 = proxy
780 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
781 .await
782 .expect("second call still gets headers");
783 let _ = drain(r2.into_body_stream()).await;
784
785 let r3 = proxy
786 .forward(&host, &nsid("sh.tangled.repo.blob"), &[], HeaderMap::new())
787 .await;
788 assert!(
789 matches!(r3, Err(KnotProxyError::CircuitOpen)),
790 "two truncated streams must trip the breaker, got {r3:?}",
791 );
792 server.abort();
793 }
794}