ptr.pet / bobbin
This repository has no description
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

refactor: thread runtime weaving to ingest, knot-proxy, slingshot-client

Lewis: May this revision serve well! <lu5a@proton.me>

Lewis (May 6, 2026, 12:54 PM +0300) 4574683a ad180b2c

+400 -229
+1 -1
crates/ingest/Cargo.toml
··· 9 9 bobbin-types = { workspace = true } 10 10 bobbin-edge-index = { workspace = true } 11 11 bobbin-record-lru = { workspace = true } 12 + bobbin-runtime = { workspace = true } 12 13 bobbin-slingshot-client = { workspace = true } 13 14 jacquard-common = { workspace = true } 14 15 ··· 19 20 serde_json = { workspace = true } 20 21 thiserror = { workspace = true } 21 22 tokio = { workspace = true } 22 - tokio-tungstenite = { workspace = true } 23 23 tokio-util = { workspace = true } 24 24 tracing = { workspace = true } 25 25 url = { workspace = true }
+7 -2
crates/ingest/examples/smoke.rs
··· 4 4 use bobbin_edge_index::{CoverageWatch, EdgeStore}; 5 5 use bobbin_ingest::{IngestConfig, IngestRuntime, RepoIdResolver, run}; 6 6 use bobbin_record_lru::{NoopRecordStore, RecordStore}; 7 + use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, TungsteniteWs}; 7 8 use bobbin_types::search::NoopSearchSink; 8 9 use futures::stream::{self, StreamExt}; 9 10 use tokio_util::sync::CancellationToken; ··· 24 25 .unwrap_or(6); 25 26 26 27 let url = Url::parse(&endpoint).expect("valid hydrant base url"); 27 - let store = Arc::new(EdgeStore::new()); 28 + let hasher = RuntimeHasher::from_entropy(&OsEntropy); 29 + let store = Arc::new(EdgeStore::new(hasher.clone())); 28 30 let coverage = Arc::new(CoverageWatch::new()); 29 31 30 32 let cfg = IngestConfig::new(url); ··· 34 36 coverage: coverage.clone(), 35 37 search: Arc::new(NoopSearchSink), 36 38 records: Arc::new(NoopRecordStore) as Arc<dyn RecordStore>, 37 - resolver: Arc::new(RepoIdResolver::detached()), 39 + resolver: Arc::new(RepoIdResolver::detached(hasher)), 40 + clock: Arc::new(SystemClock::new()), 41 + entropy: Arc::new(OsEntropy), 42 + ws: TungsteniteWs::shared(), 38 43 cancel: cancel.clone(), 39 44 }; 40 45 let task = tokio::spawn(async move {
+137 -106
crates/ingest/src/lib.rs
··· 1 1 use std::sync::Arc; 2 - use std::time::{Duration, SystemTime, UNIX_EPOCH}; 2 + use std::time::Duration; 3 3 4 4 use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, PromotionSignal}; 5 5 use bobbin_record_lru::RecordStore; 6 + use bobbin_runtime::{ 7 + Clock, Entropy, NetworkError, UnixMicros, WsConn, WsMessage, WsStream, WsTransport, 8 + }; 6 9 use bobbin_types::edges::{Edge, ExtractError, Record}; 7 10 use bobbin_types::record::RecordBody; 8 11 use bobbin_types::search::{SearchSink, SearchableRecord}; 9 - use futures::{SinkExt, StreamExt}; 12 + use bytes::Bytes; 13 + use futures::StreamExt; 10 14 use jacquard_common::DefaultStr; 11 15 use jacquard_common::types::did::Did; 12 16 use jacquard_common::types::ident::AtIdentifier; 13 17 use jacquard_common::types::recordkey::Rkey; 14 18 use jacquard_common::types::string::{AtStrError, AtUri, Cid}; 15 19 use thiserror::Error; 16 - use tokio::time::{Instant, MissedTickBehavior, interval}; 17 - use tokio_tungstenite::tungstenite::{ 18 - Bytes, Message, protocol::CloseFrame, protocol::frame::coding::CloseCode, 19 - }; 20 + use tokio::time::Instant; 20 21 use tokio_util::sync::CancellationToken; 21 22 use tracing::{debug, info, warn}; 22 23 use url::Url; ··· 35 36 const FRAME_CHANNEL_DEPTH: usize = 256; 36 37 const CONTROL_CHANNEL_DEPTH: usize = 16; 37 38 const NORMALIZE_CONCURRENCY: usize = 8; 39 + const NORMAL_CLOSE: u16 = 1000; 38 40 39 41 #[derive(Clone, Debug)] 40 42 pub struct IngestConfig { ··· 76 78 Url(&'static str), 77 79 #[error("unsupported url scheme: {0}")] 78 80 UnknownScheme(String), 79 - #[error("websocket transport: {0}")] 80 - Transport(#[from] tokio_tungstenite::tungstenite::Error), 81 + #[error("network: {0}")] 82 + Network(#[from] NetworkError), 81 83 #[error("frame decode: {0}")] 82 84 Decode(#[from] serde_json::Error), 83 85 #[error("invalid at-uri synthesized from frame: {0}")] ··· 106 108 pub search: Arc<S>, 107 109 pub records: Arc<dyn RecordStore>, 108 110 pub resolver: Arc<RepoIdResolver>, 111 + pub clock: Arc<dyn Clock>, 112 + pub entropy: Arc<dyn Entropy>, 113 + pub ws: Arc<dyn WsTransport>, 109 114 pub cancel: CancellationToken, 110 115 } 111 116 ··· 117 122 search: self.search.clone(), 118 123 records: self.records.clone(), 119 124 resolver: self.resolver.clone(), 125 + clock: self.clock.clone(), 126 + entropy: self.entropy.clone(), 127 + ws: self.ws.clone(), 120 128 cancel: self.cancel.clone(), 121 129 } 122 130 } ··· 151 159 backoff = RECONNECT_INITIAL_DELAY; 152 160 } else { 153 161 tokio::select! { 154 - _ = tokio::time::sleep(jittered(backoff)) => {} 162 + _ = runtime.clock.sleep(jittered(backoff, &*runtime.entropy)) => {} 155 163 _ = runtime.cancel.cancelled() => return Ok(()), 156 164 } 157 165 backoff = (backoff * 2).min(RECONNECT_MAX_DELAY); ··· 167 175 } 168 176 } 169 177 170 - fn jittered(base: Duration) -> Duration { 171 - let entropy = SystemTime::now() 172 - .duration_since(UNIX_EPOCH) 173 - .map(|d| d.subsec_nanos() as u64) 174 - .unwrap_or(0); 175 - let cap_ms = (base.as_millis() as u64 / 4).max(1); 176 - base + Duration::from_millis(entropy % cap_ms) 178 + fn jittered(base: Duration, entropy: &dyn Entropy) -> Duration { 179 + let base_ms = u64::try_from(base.as_millis()).unwrap_or(u64::MAX); 180 + let cap_ms = (base_ms / 4).max(1); 181 + base + Duration::from_millis(entropy.next_u64() % cap_ms) 177 182 } 178 183 179 184 async fn run_session<S: SearchSink + 'static>( ··· 196 201 _ = runtime.cancel.cancelled() => { 197 202 return SessionEnd { outcome: SessionOutcome::Empty, error: None }; 198 203 } 199 - res = tokio_tungstenite::connect_async(url.as_str()) => res, 204 + res = runtime.ws.connect(url) => res, 200 205 }; 201 - let (ws, _resp) = match connect { 202 - Ok(pair) => pair, 206 + let WsConn { 207 + sink: mut ws_sink, 208 + stream: ws_stream, 209 + } = match connect { 210 + Ok(c) => c, 203 211 Err(e) => { 204 212 return SessionEnd { 205 213 outcome: SessionOutcome::Empty, 206 - error: Some(IngestError::Transport(e)), 214 + error: Some(IngestError::Network(e)), 207 215 }; 208 216 } 209 217 }; 210 - let (mut ws_sink, mut ws_stream) = ws.split(); 211 218 212 219 let (frame_tx, mut frame_rx) = tokio::sync::mpsc::channel::<HydrantFrame>(FRAME_CHANNEL_DEPTH); 213 220 let processor_runtime = runtime.clone(); ··· 218 225 _ = processor_runtime.cancel.cancelled() => break, 219 226 next = frame_rx.recv() => { 220 227 let Some(frame) = next else { break }; 228 + let now = processor_runtime.clock.now_unix_micros(); 221 229 handle_frame( 222 230 frame, 223 231 &processor_runtime.store, ··· 225 233 &*processor_runtime.search, 226 234 &*processor_runtime.records, 227 235 &processor_runtime.resolver, 236 + now, 228 237 ) 229 238 .await; 230 239 } ··· 235 244 let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::<WsEvent>(CONTROL_CHANNEL_DEPTH); 236 245 let session_cancel = runtime.cancel.child_token(); 237 246 let reader_cancel = session_cancel.clone(); 238 - let reader = tokio::spawn(async move { 239 - let mut outcome = SessionOutcome::Empty; 240 - let error: Option<IngestError> = loop { 241 - tokio::select! { 242 - biased; 243 - _ = reader_cancel.cancelled() => break None, 244 - msg = ws_stream.next() => { 245 - let Some(msg) = msg else { break None; }; 246 - let parsed = match msg { 247 - Ok(m) => m, 248 - Err(e) => break Some(IngestError::Transport(e)), 249 - }; 250 - match parsed { 251 - Message::Text(text) => { 252 - let frame: HydrantFrame = match serde_json::from_str(&text) { 253 - Ok(f) => f, 254 - Err(e) => break Some(IngestError::Decode(e)), 255 - }; 256 - tokio::select! { 257 - biased; 258 - _ = reader_cancel.cancelled() => break None, 259 - res = frame_tx.send(frame) => { 260 - if res.is_err() { break None; } 261 - } 262 - } 263 - outcome = SessionOutcome::Progressed; 264 - } 265 - Message::Binary(_) => { 266 - debug!("hydrant sent unexpected binary frame, ignoring"); 267 - } 268 - Message::Ping(payload) => { 269 - if control_tx.send(WsEvent::IncomingPing(payload)).await.is_err() { 270 - break None; 271 - } 272 - } 273 - Message::Pong(_) => { 274 - if control_tx.send(WsEvent::IncomingPong).await.is_err() { 275 - break None; 276 - } 277 - } 278 - Message::Frame(_) => {} 279 - Message::Close(close) => { 280 - debug!(?close, "hydrant closed stream"); 281 - break None; 282 - } 283 - } 284 - } 285 - } 286 - }; 287 - SessionEnd { outcome, error } 288 - }); 247 + let reader = tokio::spawn(reader_loop(ws_stream, frame_tx, control_tx, reader_cancel)); 289 248 290 - let mut pinger = interval(PING_INTERVAL); 291 - pinger.set_missed_tick_behavior(MissedTickBehavior::Delay); 292 - pinger.tick().await; 249 + let mut next_ping = runtime.clock.now_instant() + PING_INTERVAL; 293 250 let mut pong_deadline: Option<Instant> = None; 294 251 295 252 let writer_error: Option<IngestError> = loop { 296 253 tokio::select! { 297 254 biased; 298 255 _ = runtime.cancel.cancelled() => { 299 - let close_frame = CloseFrame { code: CloseCode::Normal, reason: "bobbin shutdown".into() }; 300 - if let Err(e) = ws_sink.send(Message::Close(Some(close_frame))).await { 301 - debug!(?e, "could not send hydrant close frame on shutdown"); 302 - } 256 + let _ = ws_sink.send(WsMessage::Close { code: NORMAL_CLOSE, reason: "bobbin shutdown".to_owned() }).await; 303 257 break None; 304 258 } 305 - _ = pinger.tick() => { 259 + _ = runtime.clock.sleep_until(next_ping) => { 260 + next_ping = runtime.clock.now_instant() + PING_INTERVAL; 306 261 if pong_deadline.is_none() { 307 - if let Err(e) = ws_sink.send(Message::Ping(Bytes::new())).await { 308 - break Some(IngestError::Transport(e)); 262 + if let Err(e) = ws_sink.send(WsMessage::Ping(Bytes::new())).await { 263 + break Some(IngestError::Network(e)); 309 264 } 310 - pong_deadline = Some(Instant::now() + PONG_TIMEOUT); 265 + pong_deadline = Some(runtime.clock.now_instant() + PONG_TIMEOUT); 311 266 } 312 267 } 313 - _ = wait_until(pong_deadline) => { 268 + _ = wait_until(pong_deadline, runtime.clock.as_ref()) => { 314 269 break Some(IngestError::PongTimeout(PONG_TIMEOUT)); 315 270 } 316 271 evt = control_rx.recv() => { 317 272 let Some(evt) = evt else { break None; }; 318 273 match evt { 319 274 WsEvent::IncomingPing(payload) => { 320 - if let Err(e) = ws_sink.send(Message::Pong(payload)).await { 321 - break Some(IngestError::Transport(e)); 275 + if let Err(e) = ws_sink.send(WsMessage::Pong(payload)).await { 276 + break Some(IngestError::Network(e)); 322 277 } 323 278 } 324 279 WsEvent::IncomingPong => { ··· 344 299 outcome: reader_end.outcome, 345 300 error: writer_error.or(reader_end.error), 346 301 } 302 + } 303 + 304 + async fn reader_loop( 305 + mut ws_stream: Box<dyn WsStream>, 306 + frame_tx: tokio::sync::mpsc::Sender<HydrantFrame>, 307 + control_tx: tokio::sync::mpsc::Sender<WsEvent>, 308 + cancel: CancellationToken, 309 + ) -> SessionEnd { 310 + let mut outcome = SessionOutcome::Empty; 311 + let error: Option<IngestError> = loop { 312 + tokio::select! { 313 + biased; 314 + _ = cancel.cancelled() => break None, 315 + msg = ws_stream.next() => { 316 + let Some(msg) = msg else { break None; }; 317 + let parsed = match msg { 318 + Ok(m) => m, 319 + Err(e) => break Some(IngestError::Network(e)), 320 + }; 321 + match parsed { 322 + WsMessage::Text(text) => { 323 + let frame: HydrantFrame = match serde_json::from_str(&text) { 324 + Ok(f) => f, 325 + Err(e) => break Some(IngestError::Decode(e)), 326 + }; 327 + tokio::select! { 328 + biased; 329 + _ = cancel.cancelled() => break None, 330 + res = frame_tx.send(frame) => { 331 + if res.is_err() { break None; } 332 + } 333 + } 334 + outcome = SessionOutcome::Progressed; 335 + } 336 + WsMessage::Binary(_) => { 337 + debug!("hydrant sent unexpected binary frame, ignoring"); 338 + } 339 + WsMessage::Ping(payload) => { 340 + if control_tx.send(WsEvent::IncomingPing(payload)).await.is_err() { 341 + break None; 342 + } 343 + } 344 + WsMessage::Pong(_) => { 345 + if control_tx.send(WsEvent::IncomingPong).await.is_err() { 346 + break None; 347 + } 348 + } 349 + WsMessage::Close { code, reason } => { 350 + debug!(code, %reason, "hydrant closed stream"); 351 + break None; 352 + } 353 + } 354 + } 355 + } 356 + }; 357 + SessionEnd { outcome, error } 347 358 } 348 359 349 360 #[derive(Debug)] ··· 352 363 IncomingPong, 353 364 } 354 365 355 - async fn wait_until(deadline: Option<Instant>) { 366 + async fn wait_until(deadline: Option<Instant>, clock: &dyn Clock) { 356 367 match deadline { 357 - Some(d) => tokio::time::sleep_until(d).await, 368 + Some(d) => clock.sleep_until(d).await, 358 369 None => std::future::pending::<()>().await, 359 370 } 360 371 } ··· 366 377 search: &S, 367 378 records: &dyn RecordStore, 368 379 resolver: &RepoIdResolver, 380 + now: UnixMicros, 369 381 ) { 370 382 let cursor = HydrantCursor::new(frame.id); 371 - let signal = promotion_signal(frame.record.as_ref(), now_micros()); 383 + let signal = promotion_signal(frame.record.as_ref(), now); 372 384 coverage.update(|c| c.advance(cursor).maybe_promote(signal)); 373 385 374 386 match frame.kind { ··· 394 406 } 395 407 } 396 408 397 - fn promotion_signal(record: Option<&RecordFrame>, now_micros: u64) -> PromotionSignal { 409 + fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { 398 410 PromotionSignal { 399 411 live: record.is_some_and(|r| r.live), 400 412 rev_micros: record.map(|r| r.rev.timestamp()), 401 - now_micros, 413 + now_micros: now.raw(), 402 414 skew_micros: READY_SKEW.as_micros() as u64, 403 415 } 404 416 } 405 417 406 - fn now_micros() -> u64 { 407 - SystemTime::now() 408 - .duration_since(UNIX_EPOCH) 409 - .expect("system clock before unix epoch") 410 - .as_micros() as u64 411 - } 412 - 413 418 async fn apply_record<S: SearchSink>( 414 419 record: RecordFrame, 415 420 store: &EdgeStore, ··· 538 543 use super::*; 539 544 use bobbin_edge_index::Coverage; 540 545 use bobbin_record_lru::{CacheCapacity, LruRecordStore, NoopRecordStore, RecordStore}; 546 + use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, TungsteniteWs}; 541 547 use bobbin_types::search::NoopSearchSink; 542 548 use jacquard_common::types::nsid::Nsid; 543 549 use jacquard_common::types::tid::Tid; ··· 547 553 548 554 fn fresh() -> (Arc<EdgeStore>, Arc<CoverageWatch>, Arc<RepoIdResolver>) { 549 555 ( 550 - Arc::new(EdgeStore::new()), 556 + Arc::new(EdgeStore::new(RuntimeHasher::default())), 551 557 Arc::new(CoverageWatch::new()), 552 - Arc::new(RepoIdResolver::detached()), 558 + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), 553 559 ) 554 560 } 555 561 562 + fn now() -> UnixMicros { 563 + SystemClock::new().now_unix_micros() 564 + } 565 + 556 566 fn parse_frame(value: serde_json::Value) -> HydrantFrame { 557 567 let text = serde_json::to_string(&value).expect("serialize fixture"); 558 568 serde_json::from_str(&text).expect("deserialize fixture") ··· 585 595 &NoopSearchSink, 586 596 &NoopRecordStore, 587 597 &resolver, 598 + now(), 588 599 ) 589 600 .await; 590 601 assert_eq!(store.key_count(), 0); ··· 618 629 &NoopSearchSink, 619 630 &NoopRecordStore, 620 631 &resolver, 632 + now(), 621 633 ) 622 634 .await; 623 635 let key = bobbin_types::ids::EdgeKey::new( ··· 646 658 &NoopSearchSink, 647 659 &NoopRecordStore, 648 660 &resolver, 661 + now(), 649 662 ) 650 663 .await; 651 664 assert_eq!(store.count(&key), 0); ··· 680 693 &NoopSearchSink, 681 694 &NoopRecordStore, 682 695 &resolver, 696 + now(), 683 697 ) 684 698 .await; 685 699 handle_frame( ··· 689 703 &NoopSearchSink, 690 704 &NoopRecordStore, 691 705 &resolver, 706 + now(), 692 707 ) 693 708 .await; 694 709 ··· 727 742 })); 728 743 let source = 729 744 AtUri::new_owned("at://did:plc:olaren/sh.tangled.feed.star/abcabcabcabcz").unwrap(); 730 - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; 745 + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; 731 746 let cached = lru.get(&source).expect("hydrant cid must seed the lru"); 732 747 assert_eq!(cached.cid.as_ref(), VALID_CID); 733 748 let parsed: serde_json::Value = serde_json::from_slice(&cached.value).unwrap(); ··· 766 781 } 767 782 } 768 783 })); 769 - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; 784 + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; 770 785 assert!( 771 786 lru.get(&source).is_none(), 772 787 "missing cid means we cannot trust the body, so the lru must be cleared", ··· 801 816 "record": null 802 817 } 803 818 })); 804 - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; 819 + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; 805 820 assert!(lru.get(&source).is_none()); 806 821 } 807 822 ··· 833 848 &NoopSearchSink, 834 849 &NoopRecordStore, 835 850 &resolver, 851 + now(), 836 852 ) 837 853 .await; 838 854 assert!(cov.snapshot().is_ready()); ··· 867 883 &NoopSearchSink, 868 884 &NoopRecordStore, 869 885 &resolver, 886 + now(), 870 887 ) 871 888 .await; 872 889 assert!(!cov.snapshot().is_ready()); ··· 935 952 &NoopSearchSink, 936 953 &NoopRecordStore, 937 954 &resolver, 955 + now(), 938 956 ) 939 957 .await; 940 958 assert_eq!(store.key_count(), 0); ··· 957 975 &NoopSearchSink, 958 976 &NoopRecordStore, 959 977 &resolver, 978 + now(), 960 979 ) 961 980 .await; 962 981 assert_eq!(store.key_count(), 0); ··· 975 994 &NoopSearchSink, 976 995 &NoopRecordStore, 977 996 &resolver, 997 + now(), 978 998 ) 979 999 .await; 980 1000 assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(8)); ··· 982 1002 983 1003 fn fresh_runtime(cancel: CancellationToken) -> IngestRuntime<NoopSearchSink> { 984 1004 IngestRuntime { 985 - store: Arc::new(EdgeStore::new()), 1005 + store: Arc::new(EdgeStore::new(RuntimeHasher::default())), 986 1006 coverage: Arc::new(CoverageWatch::new()), 987 1007 search: Arc::new(NoopSearchSink), 988 1008 records: Arc::new(NoopRecordStore) as Arc<dyn RecordStore>, 989 - resolver: Arc::new(RepoIdResolver::detached()), 1009 + resolver: Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), 1010 + clock: Arc::new(SystemClock::new()), 1011 + entropy: Arc::new(OsEntropy), 1012 + ws: TungsteniteWs::shared(), 990 1013 cancel, 991 1014 } 992 1015 } ··· 1009 1032 fn jittered_stays_within_one_quarter_of_base() { 1010 1033 let base = Duration::from_secs(1); 1011 1034 let cap = base + Duration::from_millis(250); 1035 + let entropy = OsEntropy; 1012 1036 (0..50).for_each(|_| { 1013 - let j = jittered(base); 1037 + let j = jittered(base, &entropy); 1014 1038 assert!(j >= base, "jitter must not undershoot"); 1015 1039 assert!(j <= cap, "jitter must not exceed +25%, got {:?}", j); 1016 1040 }); ··· 1045 1069 &NoopSearchSink, 1046 1070 &NoopRecordStore, 1047 1071 &resolver, 1072 + now(), 1048 1073 ) 1049 1074 .await; 1050 1075 ··· 1072 1097 &NoopSearchSink, 1073 1098 &NoopRecordStore, 1074 1099 &resolver, 1100 + now(), 1075 1101 ) 1076 1102 .await; 1077 1103 ··· 1121 1147 &NoopSearchSink, 1122 1148 &NoopRecordStore, 1123 1149 &resolver, 1150 + now(), 1124 1151 ) 1125 1152 .await; 1126 1153 ··· 1173 1200 &NoopSearchSink, 1174 1201 &NoopRecordStore, 1175 1202 &resolver, 1203 + now(), 1176 1204 ) 1177 1205 .await; 1178 1206 ··· 1200 1228 &NoopSearchSink, 1201 1229 &NoopRecordStore, 1202 1230 &resolver, 1231 + now(), 1203 1232 ) 1204 1233 .await; 1205 1234 ··· 1241 1270 &NoopSearchSink, 1242 1271 &NoopRecordStore, 1243 1272 &resolver, 1273 + now(), 1244 1274 ) 1245 1275 .await; 1246 1276 let key = bobbin_types::ids::EdgeKey::new( ··· 1285 1315 &NoopSearchSink, 1286 1316 &NoopRecordStore, 1287 1317 &resolver, 1318 + now(), 1288 1319 ) 1289 1320 .await; 1290 1321 let key = bobbin_types::ids::EdgeKey::new(
+28 -22
crates/ingest/src/resolver.rs
··· 1 + use bobbin_runtime::RuntimeHasher; 1 2 use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; 2 3 use bobbin_types::edges::{ExtractError, Record}; 3 4 use bobbin_types::ids::nsid_static; ··· 61 62 } 62 63 63 64 pub struct RepoIdResolver { 64 - cache: SccMap<RepoRef, CacheEntry>, 65 + cache: SccMap<RepoRef, CacheEntry, RuntimeHasher>, 65 66 client: Option<SlingshotClient>, 66 67 } 67 68 68 69 impl RepoIdResolver { 69 - pub fn with_slingshot(client: SlingshotClient) -> Self { 70 + pub fn with_slingshot(client: SlingshotClient, hasher: RuntimeHasher) -> Self { 70 71 Self { 71 - cache: SccMap::new(), 72 + cache: SccMap::with_hasher(hasher), 72 73 client: Some(client), 73 74 } 74 75 } 75 76 76 - pub fn detached() -> Self { 77 + pub fn detached(hasher: RuntimeHasher) -> Self { 77 78 Self { 78 - cache: SccMap::new(), 79 + cache: SccMap::with_hasher(hasher), 79 80 client: None, 80 81 } 81 82 } ··· 197 198 198 199 #[tokio::test] 199 200 async fn observation_with_repo_did_resolves_mapped() { 200 - let resolver = RepoIdResolver::detached(); 201 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 201 202 resolver 202 203 .observe( 203 204 did("did:plc:nel"), ··· 213 214 214 215 #[tokio::test] 215 216 async fn observation_without_repo_did_resolves_no_repo_did() { 216 - let resolver = RepoIdResolver::detached(); 217 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 217 218 resolver 218 219 .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None) 219 220 .await; ··· 229 230 230 231 #[tokio::test] 231 232 async fn cache_miss_without_client_is_unresolvable() { 232 - let resolver = RepoIdResolver::detached(); 233 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 233 234 let got = resolver 234 235 .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) 235 236 .await; ··· 238 239 239 240 #[tokio::test] 240 241 async fn observation_overwrites_prior_value() { 241 - let resolver = RepoIdResolver::detached(); 242 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 242 243 resolver 243 244 .observe( 244 245 did("did:plc:nel"), ··· 261 262 262 263 #[tokio::test] 263 264 async fn fill_provisional_does_not_downgrade_authoritative_mapped() { 264 - let resolver = RepoIdResolver::detached(); 265 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 265 266 let owner = did("did:plc:nel"); 266 267 let key = rkey("abcabcabcabcz"); 267 268 resolver ··· 283 284 284 285 #[tokio::test] 285 286 async fn fill_provisional_does_not_downgrade_authoritative_no_repo_did() { 286 - let resolver = RepoIdResolver::detached(); 287 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 287 288 let owner = did("did:plc:nel"); 288 289 let key = rkey("abcabcabcabcz"); 289 290 resolver.observe(owner.clone(), key.clone(), None).await; ··· 311 312 .mount(&server) 312 313 .await; 313 314 314 - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); 315 - let resolver = RepoIdResolver::with_slingshot(client); 315 + let client = 316 + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); 317 + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); 316 318 317 319 let owner = did("did:plc:nel"); 318 320 let key = rkey("abcabcabcabcz"); ··· 336 338 .mount(&server) 337 339 .await; 338 340 339 - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); 340 - let resolver = RepoIdResolver::with_slingshot(client); 341 + let client = 342 + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); 343 + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); 341 344 342 345 let owner = did("did:plc:nel"); 343 346 let key = rkey("abcabcabcabcz"); ··· 366 369 .mount(&server) 367 370 .await; 368 371 369 - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); 370 - let resolver = RepoIdResolver::with_slingshot(client); 372 + let client = 373 + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); 374 + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); 371 375 372 376 let owner = did("did:plc:nel"); 373 377 let key = rkey("abcabcabcabcz"); ··· 392 396 .mount(&server) 393 397 .await; 394 398 395 - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); 396 - let resolver = RepoIdResolver::with_slingshot(client); 399 + let client = 400 + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); 401 + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); 397 402 398 403 let owner = did("did:plc:nel"); 399 404 let key = rkey("abcabcabcabcz"); ··· 417 422 .mount(&server) 418 423 .await; 419 424 420 - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); 421 - let resolver = RepoIdResolver::with_slingshot(client); 425 + let client = 426 + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); 427 + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); 422 428 423 429 let owner = did("did:plc:nel"); 424 430 let key = rkey("abcabcabcabcz"); ··· 430 436 431 437 #[tokio::test] 432 438 async fn firehose_observe_can_demote_provisional() { 433 - let resolver = RepoIdResolver::detached(); 439 + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); 434 440 let owner = did("did:plc:nel"); 435 441 let key = rkey("abcabcabcabcz"); 436 442 resolver
+1
crates/knot-proxy/Cargo.toml
··· 6 6 rust-version.workspace = true 7 7 8 8 [dependencies] 9 + bobbin-runtime = { workspace = true } 9 10 bytes = { workspace = true } 10 11 futures = { workspace = true } 11 12 http = { workspace = true }
+20 -5
crates/knot-proxy/src/breaker.rs
··· 1 1 use std::sync::{Arc, Mutex}; 2 - use std::time::{Duration, Instant}; 2 + use std::time::Duration; 3 3 4 + use bobbin_runtime::Clock; 4 5 use thiserror::Error; 6 + use tokio::time::Instant; 5 7 6 8 #[derive(Clone, Copy, Debug, Eq, PartialEq)] 7 9 pub struct FailureThreshold(u32); ··· 34 36 #[error("circuit breaker open")] 35 37 pub struct CircuitOpen; 36 38 37 - #[derive(Debug)] 38 39 pub struct Breaker { 39 40 state: Mutex<BreakerState>, 40 41 threshold: FailureThreshold, 41 42 cooldown: Duration, 43 + clock: Arc<dyn Clock>, 44 + } 45 + 46 + impl std::fmt::Debug for Breaker { 47 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 48 + f.debug_struct("Breaker") 49 + .field("state", &self.state) 50 + .field("threshold", &self.threshold) 51 + .field("cooldown", &self.cooldown) 52 + .finish_non_exhaustive() 53 + } 42 54 } 43 55 44 56 impl Breaker { 45 - pub fn new(threshold: FailureThreshold, cooldown: Duration) -> Self { 57 + pub fn new(threshold: FailureThreshold, cooldown: Duration, clock: Arc<dyn Clock>) -> Self { 46 58 Self { 47 59 state: Mutex::new(BreakerState::Closed { failures: 0 }), 48 60 threshold, 49 61 cooldown, 62 + clock, 50 63 } 51 64 } 52 65 53 66 pub fn try_acquire(self: &Arc<Self>) -> Result<BreakerPermit, CircuitOpen> { 54 - self.try_acquire_at(Instant::now()) 67 + self.try_acquire_at(self.clock.now_instant()) 55 68 } 56 69 57 70 fn try_acquire_at(self: &Arc<Self>, now: Instant) -> Result<BreakerPermit, CircuitOpen> { ··· 72 85 } 73 86 74 87 pub fn record_failure(&self) { 75 - self.record_failure_at(Instant::now()); 88 + self.record_failure_at(self.clock.now_instant()); 76 89 } 77 90 78 91 fn record_failure_at(&self, now: Instant) { ··· 134 147 #[cfg(test)] 135 148 mod tests { 136 149 use super::*; 150 + use bobbin_runtime::SystemClock; 137 151 138 152 fn breaker(threshold: u32, cooldown_ms: u64) -> Arc<Breaker> { 139 153 Arc::new(Breaker::new( 140 154 FailureThreshold::new(threshold).unwrap(), 141 155 Duration::from_millis(cooldown_ms), 156 + Arc::new(SystemClock::new()), 142 157 )) 143 158 } 144 159
+136 -58
crates/knot-proxy/src/lib.rs
··· 3 3 use std::task::{Context, Poll}; 4 4 use std::time::Duration; 5 5 6 + use bobbin_runtime::{ 7 + BodyStream as InnerBodyStream, Clock, HttpRequest, HttpResponseHead, HttpTransport, 8 + NetworkError, ReqwestHttp, RuntimeHasher, 9 + }; 6 10 use bytes::Bytes; 7 11 use futures::Stream; 8 - use http::HeaderMap; 9 - use reqwest::{Client, Response, StatusCode, redirect::Policy}; 12 + use http::{HeaderMap, StatusCode}; 13 + use reqwest::{Client, redirect::Policy}; 10 14 use scc::HashMap as SccMap; 11 15 use thiserror::Error; 12 16 use url::Url; ··· 25 29 pub struct KnotProxyConfig { 26 30 pub failure_threshold: FailureThreshold, 27 31 pub cooldown: Duration, 28 - pub connect_timeout: Duration, 29 - pub read_timeout: Duration, 30 32 pub allow_private_hosts: bool, 31 33 pub require_https: bool, 32 34 } ··· 36 38 Self { 37 39 failure_threshold: FailureThreshold::new(5).expect("nonzero literal"), 38 40 cooldown: Duration::from_secs(30), 41 + allow_private_hosts: false, 42 + require_https: true, 43 + } 44 + } 45 + } 46 + 47 + #[derive(Clone, Copy, Debug)] 48 + pub struct KnotHttpConfig { 49 + pub connect_timeout: Duration, 50 + pub read_timeout: Duration, 51 + } 52 + 53 + impl Default for KnotHttpConfig { 54 + fn default() -> Self { 55 + Self { 39 56 connect_timeout: Duration::from_secs(5), 40 57 read_timeout: Duration::from_secs(60), 41 - allow_private_hosts: false, 42 - require_https: true, 43 58 } 44 59 } 45 60 } ··· 68 83 } 69 84 70 85 pub struct KnotProxy { 71 - http: Client, 72 - breakers: SccMap<KnotHost, Arc<Breaker>>, 86 + http: Arc<dyn HttpTransport>, 87 + breakers: SccMap<KnotHost, Arc<Breaker>, RuntimeHasher>, 73 88 threshold: FailureThreshold, 74 89 cooldown: Duration, 75 90 allow_private_hosts: bool, 76 91 require_https: bool, 92 + clock: Arc<dyn Clock>, 77 93 } 78 94 79 95 impl KnotProxy { 80 - pub fn new(config: KnotProxyConfig) -> Result<Self, reqwest::Error> { 96 + pub fn new( 97 + config: KnotProxyConfig, 98 + http: KnotHttpConfig, 99 + clock: Arc<dyn Clock>, 100 + hasher: RuntimeHasher, 101 + ) -> Result<Self, reqwest::Error> { 81 102 let resolver = Arc::new(dns::PrivateAddressFilter::new(config.allow_private_hosts)); 82 - let http = Client::builder() 103 + let client = Client::builder() 83 104 .user_agent(USER_AGENT) 84 - .connect_timeout(config.connect_timeout) 85 - .read_timeout(config.read_timeout) 105 + .connect_timeout(http.connect_timeout) 106 + .read_timeout(http.read_timeout) 86 107 .redirect(Policy::none()) 87 108 .no_gzip() 88 109 .no_brotli() 89 110 .no_deflate() 90 111 .dns_resolver(resolver) 91 112 .build()?; 92 - Ok(Self { 113 + Ok(Self::with_transport( 114 + ReqwestHttp::shared(client), 115 + config, 116 + clock, 117 + hasher, 118 + )) 119 + } 120 + 121 + pub fn with_transport( 122 + http: Arc<dyn HttpTransport>, 123 + config: KnotProxyConfig, 124 + clock: Arc<dyn Clock>, 125 + hasher: RuntimeHasher, 126 + ) -> Self { 127 + Self { 93 128 http, 94 - breakers: SccMap::new(), 129 + breakers: SccMap::with_hasher(hasher), 95 130 threshold: config.failure_threshold, 96 131 cooldown: config.cooldown, 97 132 allow_private_hosts: config.allow_private_hosts, 98 133 require_https: config.require_https, 99 - }) 134 + clock, 135 + } 100 136 } 101 137 102 138 pub fn allows_private_hosts(&self) -> bool { ··· 120 156 .try_acquire() 121 157 .map_err(|_: CircuitOpen| KnotProxyError::CircuitOpen)?; 122 158 let url = build_xrpc_url(host, nsid, query); 123 - let outcome = self.http.get(url).headers(headers).send().await; 159 + let outcome = self.http.execute(HttpRequest { url, headers }).await; 124 160 classify(outcome, permit) 125 161 } 126 162 ··· 148 184 let entry = self.breakers.entry_async(host.clone()).await; 149 185 Arc::clone( 150 186 entry 151 - .or_insert_with(|| Arc::new(Breaker::new(self.threshold, self.cooldown))) 187 + .or_insert_with(|| { 188 + Arc::new(Breaker::new( 189 + self.threshold, 190 + self.cooldown, 191 + self.clock.clone(), 192 + )) 193 + }) 152 194 .get(), 153 195 ) 154 196 } 155 197 } 156 198 157 - #[derive(Debug)] 158 199 pub struct ProxyResponse { 159 - inner: Response, 200 + status: StatusCode, 201 + headers: HeaderMap, 202 + body: InnerBodyStream, 160 203 permit: BreakerPermit, 161 204 } 162 205 206 + impl std::fmt::Debug for ProxyResponse { 207 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 208 + f.debug_struct("ProxyResponse") 209 + .field("status", &self.status) 210 + .field("headers", &self.headers) 211 + .finish_non_exhaustive() 212 + } 213 + } 214 + 163 215 impl ProxyResponse { 164 216 pub fn status(&self) -> StatusCode { 165 - self.inner.status() 217 + self.status 166 218 } 167 219 168 220 pub fn headers(&self) -> &HeaderMap { 169 - self.inner.headers() 221 + &self.headers 170 222 } 171 223 172 224 pub fn into_body_stream(self) -> BodyStream { 173 - BodyStream::new(self.inner, self.permit) 225 + BodyStream::new(self.body, self.permit) 174 226 } 175 227 } 176 228 177 - type ChunkStream = Pin<Box<dyn Stream<Item = Result<Bytes, reqwest::Error>> + Send>>; 178 - 179 229 pub struct BodyStream { 180 - inner: ChunkStream, 230 + inner: InnerBodyStream, 181 231 permit: Option<BreakerPermit>, 182 232 } 183 233 184 234 impl BodyStream { 185 - fn new(response: Response, permit: BreakerPermit) -> Self { 235 + fn new(inner: InnerBodyStream, permit: BreakerPermit) -> Self { 186 236 Self { 187 - inner: Box::pin(response.bytes_stream()), 237 + inner, 188 238 permit: Some(permit), 189 239 } 190 240 } ··· 201 251 } 202 252 203 253 impl Stream for BodyStream { 204 - type Item = Result<Bytes, reqwest::Error>; 254 + type Item = Result<Bytes, NetworkError>; 205 255 206 256 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { 207 257 let next = match self.inner.as_mut().poll_next(cx) { ··· 229 279 } 230 280 231 281 fn classify( 232 - outcome: Result<Response, reqwest::Error>, 282 + outcome: Result<HttpResponseHead, NetworkError>, 233 283 permit: BreakerPermit, 234 284 ) -> Result<ProxyResponse, KnotProxyError> { 235 285 match outcome { 236 - Ok(resp) if is_upstream_failure(resp.status()) => { 237 - let status = resp.status(); 286 + Ok(head) if is_upstream_failure(head.status) => { 287 + let status = head.status; 238 288 permit.record_failure(); 239 289 Err(KnotProxyError::Upstream(status)) 240 290 } 241 - Ok(resp) => Ok(ProxyResponse { 242 - inner: resp, 291 + Ok(head) => Ok(ProxyResponse { 292 + status: head.status, 293 + headers: head.headers, 294 + body: head.body, 243 295 permit, 244 296 }), 245 297 Err(err) => { 246 298 permit.record_failure(); 247 - Err(map_transport(err)) 299 + Err(map_network(err)) 248 300 } 249 301 } 250 302 } ··· 257 309 matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308) 258 310 } 259 311 260 - fn map_transport(err: reqwest::Error) -> KnotProxyError { 261 - let msg = err.to_string(); 262 - if err.is_timeout() { 263 - KnotProxyError::Timeout(msg) 264 - } else if err.is_connect() { 265 - KnotProxyError::Connect(msg) 266 - } else if err.is_redirect() { 267 - KnotProxyError::Redirect(msg) 268 - } else { 269 - KnotProxyError::Transport(msg) 312 + fn map_network(err: NetworkError) -> KnotProxyError { 313 + match err { 314 + NetworkError::Timeout(msg) => KnotProxyError::Timeout(msg), 315 + NetworkError::Connect(msg) => KnotProxyError::Connect(msg), 316 + NetworkError::Redirect(msg) => KnotProxyError::Redirect(msg), 317 + NetworkError::Transport(msg) | NetworkError::Body(msg) | NetworkError::Protocol(msg) => { 318 + KnotProxyError::Transport(msg) 319 + } 270 320 } 271 321 } 272 322 273 323 #[cfg(test)] 274 324 mod tests { 275 325 use super::*; 326 + use bobbin_runtime::SystemClock; 276 327 use futures::stream::TryStreamExt; 277 328 use tokio::io::AsyncWriteExt; 278 329 use wiremock::matchers::{method, path, query_param}; ··· 282 333 KnotProxyConfig { 283 334 failure_threshold: FailureThreshold::new(2).unwrap(), 284 335 cooldown: Duration::from_millis(80), 336 + allow_private_hosts: true, 337 + require_https: false, 338 + } 339 + } 340 + 341 + pub(crate) fn http_config_for_test() -> KnotHttpConfig { 342 + KnotHttpConfig { 285 343 connect_timeout: Duration::from_millis(500), 286 344 read_timeout: Duration::from_secs(2), 287 - allow_private_hosts: true, 288 - require_https: false, 289 345 } 346 + } 347 + 348 + fn proxy_for_test() -> KnotProxy { 349 + KnotProxy::new( 350 + config_for_test(), 351 + http_config_for_test(), 352 + Arc::new(SystemClock::new()), 353 + RuntimeHasher::default(), 354 + ) 355 + .unwrap() 290 356 } 291 357 292 358 async fn server() -> MockServer { ··· 297 363 KnotHost::parse(&server.uri()).unwrap() 298 364 } 299 365 300 - pub(crate) async fn drain(stream: BodyStream) -> Result<Bytes, reqwest::Error> { 366 + pub(crate) async fn drain(stream: BodyStream) -> Result<Bytes, NetworkError> { 301 367 let chunks: Vec<Bytes> = stream.try_collect().await?; 302 368 let total: usize = chunks.iter().map(|b| b.len()).sum(); 303 369 let mut buf = bytes::BytesMut::with_capacity(total); ··· 321 387 .mount(&server) 322 388 .await; 323 389 324 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 390 + let proxy = proxy_for_test(); 325 391 let resp = proxy 326 392 .forward( 327 393 &host_of(&server), ··· 349 415 .mount(&server) 350 416 .await; 351 417 352 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 418 + let proxy = proxy_for_test(); 353 419 let host = host_of(&server); 354 420 let r1 = proxy 355 421 .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) ··· 374 440 .mount(&server) 375 441 .await; 376 442 377 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 443 + let proxy = proxy_for_test(); 378 444 let host = host_of(&server); 379 445 let r1 = proxy 380 446 .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) ··· 413 479 .mount(&server) 414 480 .await; 415 481 416 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 482 + let proxy = proxy_for_test(); 417 483 let host = host_of(&server); 418 484 let _ = proxy 419 485 .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) ··· 456 522 .mount(&good) 457 523 .await; 458 524 459 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 525 + let proxy = proxy_for_test(); 460 526 let bad_host = host_of(&bad); 461 527 let good_host = host_of(&good); 462 528 let _ = proxy ··· 499 565 allow_private_hosts: false, 500 566 ..config_for_test() 501 567 }; 502 - let proxy = KnotProxy::new(strict).unwrap(); 568 + let proxy = KnotProxy::new( 569 + strict, 570 + http_config_for_test(), 571 + Arc::new(SystemClock::new()), 572 + RuntimeHasher::default(), 573 + ) 574 + .unwrap(); 503 575 let err = proxy 504 576 .forward( 505 577 &host_of(&server), ··· 519 591 require_https: true, 520 592 ..config_for_test() 521 593 }; 522 - let proxy = KnotProxy::new(strict).unwrap(); 594 + let proxy = KnotProxy::new( 595 + strict, 596 + http_config_for_test(), 597 + Arc::new(SystemClock::new()), 598 + RuntimeHasher::default(), 599 + ) 600 + .unwrap(); 523 601 let err = proxy 524 602 .forward( 525 603 &host_of(&server), ··· 542 620 drop(listener); 543 621 let dead = KnotHost::parse(&format!("http://{addr}")).unwrap(); 544 622 545 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 623 + let proxy = proxy_for_test(); 546 624 let r1 = proxy 547 625 .forward(&dead, "sh.tangled.repo.blob", &[], HeaderMap::new()) 548 626 .await; ··· 581 659 .mount(&secondary) 582 660 .await; 583 661 584 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 662 + let proxy = proxy_for_test(); 585 663 let err = proxy 586 664 .forward( 587 665 &host_of(&primary), ··· 607 685 .respond_with(ResponseTemplate::new(304).insert_header("etag", "\"v1\"")) 608 686 .mount(&server) 609 687 .await; 610 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 688 + let proxy = proxy_for_test(); 611 689 let resp = proxy 612 690 .forward( 613 691 &host_of(&server), ··· 643 721 }); 644 722 645 723 let host = KnotHost::parse(&format!("http://{addr}")).unwrap(); 646 - let proxy = KnotProxy::new(config_for_test()).unwrap(); 724 + let proxy = proxy_for_test(); 647 725 648 726 let r1 = proxy 649 727 .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new())
+2
crates/slingshot-client/Cargo.toml
··· 6 6 rust-version.workspace = true 7 7 8 8 [dependencies] 9 + bobbin-runtime = { workspace = true } 9 10 bobbin-types = { workspace = true } 10 11 jacquard-common = { workspace = true } 11 12 12 13 bytes = { workspace = true } 13 14 cid = { workspace = true } 14 15 futures = { workspace = true } 16 + http = { workspace = true } 15 17 reqwest = { workspace = true } 16 18 serde = { workspace = true } 17 19 serde_json = { workspace = true }
+68 -35
crates/slingshot-client/src/lib.rs
··· 1 1 use std::sync::Arc; 2 2 use std::time::Duration; 3 3 4 + use bobbin_runtime::{HttpRequest, HttpResponseHead, HttpTransport, NetworkError, ReqwestHttp}; 4 5 use bobbin_types::record::RecordBody; 5 6 use bytes::{Bytes, BytesMut}; 6 7 use cid::Cid as IpldCid; 7 8 use futures::TryStreamExt; 9 + use http::{HeaderMap, StatusCode}; 8 10 use jacquard_common::BosStr; 9 11 use jacquard_common::types::did::Did; 10 12 use jacquard_common::types::ident::AtIdentifier; 11 13 use jacquard_common::types::nsid::Nsid; 12 14 use jacquard_common::types::recordkey::Rkey; 13 15 use jacquard_common::types::string::{AtStrError, AtUri}; 14 - use reqwest::{Client, Response, StatusCode}; 15 16 use serde::Deserialize; 16 17 use serde_json::value::RawValue; 17 18 use thiserror::Error; ··· 24 25 const RESOLVE_MINI_DOC_PATH: &str = "xrpc/com.bad-example.identity.resolveMiniDoc"; 25 26 pub const MAX_BODY_BYTES: u64 = 4 * 1024 * 1024; 26 27 27 - #[derive(Clone, Debug)] 28 + #[derive(Clone)] 28 29 pub struct SlingshotClient { 29 - http: Client, 30 + http: Arc<dyn HttpTransport>, 30 31 base: Url, 31 32 } 32 33 34 + impl std::fmt::Debug for SlingshotClient { 35 + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { 36 + f.debug_struct("SlingshotClient") 37 + .field("base", &self.base) 38 + .finish_non_exhaustive() 39 + } 40 + } 41 + 33 42 #[derive(Debug, Error)] 34 43 pub enum SlingshotError { 35 44 #[error("invalid base url scheme: {0}")] 36 45 BadScheme(String), 37 - #[error("transport: {0}")] 38 - Transport(#[from] reqwest::Error), 46 + #[error("network: {0}")] 47 + Network(#[from] NetworkError), 48 + #[error("http client build: {0}")] 49 + Build(String), 39 50 #[error("record not found")] 40 51 NotFound, 41 52 #[error("upstream returned status {0}")] ··· 54 65 UriMismatch { expected: String, got: String }, 55 66 } 56 67 68 + pub fn default_http_client() -> Result<reqwest::Client, reqwest::Error> { 69 + reqwest::Client::builder() 70 + .user_agent(USER_AGENT) 71 + .timeout(REQUEST_TIMEOUT) 72 + .connect_timeout(CONNECT_TIMEOUT) 73 + .build() 74 + } 75 + 57 76 impl SlingshotClient { 58 - pub fn new(base: Url) -> Result<Self, SlingshotError> { 77 + pub fn new(base: Url, http: Arc<dyn HttpTransport>) -> Result<Self, SlingshotError> { 59 78 match base.scheme() { 60 79 "http" | "https" => {} 61 80 other => return Err(SlingshotError::BadScheme(other.to_owned())), 62 81 } 63 82 let base = ensure_trailing_slash(base); 64 - let http = Client::builder() 65 - .user_agent(USER_AGENT) 66 - .timeout(REQUEST_TIMEOUT) 67 - .connect_timeout(CONNECT_TIMEOUT) 68 - .build()?; 69 83 Ok(Self { http, base }) 70 84 } 71 85 86 + pub fn with_default_http(base: Url) -> Result<Self, SlingshotError> { 87 + let client = default_http_client().map_err(|e| SlingshotError::Build(e.to_string()))?; 88 + Self::new(base, ReqwestHttp::shared(client)) 89 + } 90 + 72 91 pub async fn resolve_mini_doc<S>( 73 92 &self, 74 93 identifier: &AtIdentifier<S>, ··· 83 102 .clear() 84 103 .append_pair("identifier", identifier.as_str()); 85 104 86 - let resp = self.http.get(url).send().await?; 87 - match resp.status() { 105 + let resp = self 106 + .http 107 + .execute(HttpRequest { 108 + url, 109 + headers: HeaderMap::new(), 110 + }) 111 + .await?; 112 + match resp.status { 88 113 StatusCode::OK => read_bounded(resp).await, 89 114 StatusCode::NOT_FOUND => Err(SlingshotError::NotFound), 90 115 other => Err(SlingshotError::Upstream(other)), ··· 110 135 .append_pair("collection", collection.as_ref()) 111 136 .append_pair("rkey", rkey.as_ref()); 112 137 113 - let resp = self.http.get(url).send().await?; 114 - match resp.status() { 138 + let resp = self 139 + .http 140 + .execute(HttpRequest { 141 + url, 142 + headers: HeaderMap::new(), 143 + }) 144 + .await?; 145 + match resp.status { 115 146 StatusCode::OK => { 116 147 let bytes = read_bounded(resp).await?; 117 148 let body = decode(&bytes)?; ··· 132 163 base 133 164 } 134 165 135 - async fn read_bounded(resp: Response) -> Result<Bytes, SlingshotError> { 136 - if resp 137 - .content_length() 138 - .is_some_and(|len| len > MAX_BODY_BYTES) 139 - { 166 + async fn read_bounded(resp: HttpResponseHead) -> Result<Bytes, SlingshotError> { 167 + if resp.content_length.is_some_and(|len| len > MAX_BODY_BYTES) { 140 168 return Err(SlingshotError::BodyTooLarge { 141 169 limit: MAX_BODY_BYTES, 142 170 }); 143 171 } 144 172 let buf = resp 145 - .bytes_stream() 146 - .map_err(SlingshotError::Transport) 173 + .body 174 + .map_err(SlingshotError::Network) 147 175 .try_fold(BytesMut::new(), |mut acc, chunk| async move { 148 176 if (acc.len() as u64).saturating_add(chunk.len() as u64) > MAX_BODY_BYTES { 149 177 return Err(SlingshotError::BodyTooLarge { ··· 217 245 MockServer::start().await 218 246 } 219 247 248 + fn client_for(server: &MockServer) -> SlingshotClient { 249 + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap() 250 + } 251 + 220 252 #[tokio::test] 221 253 async fn returns_decoded_record_on_200() { 222 254 let server = server().await; ··· 238 270 .mount(&server) 239 271 .await; 240 272 241 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 273 + let client = client_for(&server); 242 274 let resp = client 243 275 .get_record( 244 276 &did("did:plc:abalone"), ··· 268 300 .mount(&server) 269 301 .await; 270 302 271 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 303 + let client = client_for(&server); 272 304 let resp = client 273 305 .get_record(&did("did:plc:uni"), &nsid("sh.tangled.repo"), &rkey("r1")) 274 306 .await ··· 285 317 .mount(&server) 286 318 .await; 287 319 288 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 320 + let client = client_for(&server); 289 321 let err = client 290 322 .get_record( 291 323 &did("did:plc:abalone"), ··· 306 338 .mount(&server) 307 339 .await; 308 340 309 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 341 + let client = client_for(&server); 310 342 let err = client 311 343 .get_record( 312 344 &did("did:plc:abalone"), ··· 330 362 .mount(&server) 331 363 .await; 332 364 333 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 365 + let client = client_for(&server); 334 366 let err = client 335 367 .get_record( 336 368 &did("did:plc:abalone"), ··· 359 391 .mount(&server) 360 392 .await; 361 393 362 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 394 + let client = client_for(&server); 363 395 let err = client 364 396 .get_record( 365 397 &did("did:plc:abalone"), ··· 388 420 .mount(&server) 389 421 .await; 390 422 391 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 423 + let client = client_for(&server); 392 424 let err = client 393 425 .get_record( 394 426 &did("did:plc:abalone"), ··· 417 449 .mount(&server) 418 450 .await; 419 451 420 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 452 + let client = client_for(&server); 421 453 let err = client 422 454 .get_record( 423 455 &did("did:plc:abalone"), ··· 449 481 .mount(&server) 450 482 .await; 451 483 452 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 484 + let client = client_for(&server); 453 485 let err = client 454 486 .get_record( 455 487 &did("did:plc:abalone"), ··· 477 509 .mount(&server) 478 510 .await; 479 511 480 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 512 + let client = client_for(&server); 481 513 let err = client 482 514 .get_record( 483 515 &did("did:plc:abalone"), ··· 494 526 495 527 #[test] 496 528 fn rejects_non_http_scheme() { 497 - let err = SlingshotClient::new(Url::parse("ftp://nel.pet").unwrap()).expect_err("ftp bad"); 529 + let err = SlingshotClient::with_default_http(Url::parse("ftp://nel.pet").unwrap()) 530 + .expect_err("ftp bad"); 498 531 assert!(matches!(err, SlingshotError::BadScheme(s) if s == "ftp")); 499 532 } 500 533 ··· 518 551 .await; 519 552 520 553 let base = Url::parse(&format!("{}/api/v0", server.uri())).unwrap(); 521 - let client = SlingshotClient::new(base).unwrap(); 554 + let client = SlingshotClient::with_default_http(base).unwrap(); 522 555 client 523 556 .get_record( 524 557 &did("did:plc:abalone"), ··· 555 588 jacquard_common::types::ident::AtIdentifier::Handle(_) => unreachable!(), 556 589 }; 557 590 558 - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); 591 + let client = client_for(&server); 559 592 let resp = client 560 593 .get_record(&did_borrow, &collection, &rkey) 561 594 .await