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