From 4574683a0da77d3b8ef97a317eda833ca8d1ee1e Mon Sep 17 00:00:00 2001 From: Lewis Date: Wed, 6 May 2026 12:54:17 +0300 Subject: [PATCH] refactor: thread runtime weaving to ingest, knot-proxy, slingshot-client Lewis: May this revision serve well! --- crates/ingest/Cargo.toml | 2 +- crates/ingest/examples/smoke.rs | 9 +- crates/ingest/src/lib.rs | 243 ++++++++++++++++------------- crates/ingest/src/resolver.rs | 50 +++--- crates/knot-proxy/Cargo.toml | 1 + crates/knot-proxy/src/breaker.rs | 25 ++- crates/knot-proxy/src/lib.rs | 194 ++++++++++++++++------- crates/slingshot-client/Cargo.toml | 2 + crates/slingshot-client/src/lib.rs | 103 +++++++----- 9 files changed, 400 insertions(+), 229 deletions(-) diff --git a/crates/ingest/Cargo.toml b/crates/ingest/Cargo.toml index 1ef2e8c..f3c49b9 100644 --- a/crates/ingest/Cargo.toml +++ b/crates/ingest/Cargo.toml @@ -9,6 +9,7 @@ rust-version.workspace = true bobbin-types = { workspace = true } bobbin-edge-index = { workspace = true } bobbin-record-lru = { workspace = true } +bobbin-runtime = { workspace = true } bobbin-slingshot-client = { workspace = true } jacquard-common = { workspace = true } @@ -19,7 +20,6 @@ serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } tokio = { workspace = true } -tokio-tungstenite = { workspace = true } tokio-util = { workspace = true } tracing = { workspace = true } url = { workspace = true } diff --git a/crates/ingest/examples/smoke.rs b/crates/ingest/examples/smoke.rs index a5beeeb..c2ebcd6 100644 --- a/crates/ingest/examples/smoke.rs +++ b/crates/ingest/examples/smoke.rs @@ -4,6 +4,7 @@ use std::time::Duration; use bobbin_edge_index::{CoverageWatch, EdgeStore}; use bobbin_ingest::{IngestConfig, IngestRuntime, RepoIdResolver, run}; use bobbin_record_lru::{NoopRecordStore, RecordStore}; +use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, TungsteniteWs}; use bobbin_types::search::NoopSearchSink; use futures::stream::{self, StreamExt}; use tokio_util::sync::CancellationToken; @@ -24,7 +25,8 @@ async fn main() { .unwrap_or(6); let url = Url::parse(&endpoint).expect("valid hydrant base url"); - let store = Arc::new(EdgeStore::new()); + let hasher = RuntimeHasher::from_entropy(&OsEntropy); + let store = Arc::new(EdgeStore::new(hasher.clone())); let coverage = Arc::new(CoverageWatch::new()); let cfg = IngestConfig::new(url); @@ -34,7 +36,10 @@ async fn main() { coverage: coverage.clone(), search: Arc::new(NoopSearchSink), records: Arc::new(NoopRecordStore) as Arc, - resolver: Arc::new(RepoIdResolver::detached()), + resolver: Arc::new(RepoIdResolver::detached(hasher)), + clock: Arc::new(SystemClock::new()), + entropy: Arc::new(OsEntropy), + ws: TungsteniteWs::shared(), cancel: cancel.clone(), }; let task = tokio::spawn(async move { diff --git a/crates/ingest/src/lib.rs b/crates/ingest/src/lib.rs index bc819a1..713b4d1 100644 --- a/crates/ingest/src/lib.rs +++ b/crates/ingest/src/lib.rs @@ -1,22 +1,23 @@ use std::sync::Arc; -use std::time::{Duration, SystemTime, UNIX_EPOCH}; +use std::time::Duration; use bobbin_edge_index::{Coverage, CoverageWatch, EdgeStore, HydrantCursor, PromotionSignal}; use bobbin_record_lru::RecordStore; +use bobbin_runtime::{ + Clock, Entropy, NetworkError, UnixMicros, WsConn, WsMessage, WsStream, WsTransport, +}; use bobbin_types::edges::{Edge, ExtractError, Record}; use bobbin_types::record::RecordBody; use bobbin_types::search::{SearchSink, SearchableRecord}; -use futures::{SinkExt, StreamExt}; +use bytes::Bytes; +use futures::StreamExt; use jacquard_common::DefaultStr; use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::{AtStrError, AtUri, Cid}; use thiserror::Error; -use tokio::time::{Instant, MissedTickBehavior, interval}; -use tokio_tungstenite::tungstenite::{ - Bytes, Message, protocol::CloseFrame, protocol::frame::coding::CloseCode, -}; +use tokio::time::Instant; use tokio_util::sync::CancellationToken; use tracing::{debug, info, warn}; use url::Url; @@ -35,6 +36,7 @@ const READY_SKEW: Duration = Duration::from_secs(60); const FRAME_CHANNEL_DEPTH: usize = 256; const CONTROL_CHANNEL_DEPTH: usize = 16; const NORMALIZE_CONCURRENCY: usize = 8; +const NORMAL_CLOSE: u16 = 1000; #[derive(Clone, Debug)] pub struct IngestConfig { @@ -76,8 +78,8 @@ pub enum IngestError { Url(&'static str), #[error("unsupported url scheme: {0}")] UnknownScheme(String), - #[error("websocket transport: {0}")] - Transport(#[from] tokio_tungstenite::tungstenite::Error), + #[error("network: {0}")] + Network(#[from] NetworkError), #[error("frame decode: {0}")] Decode(#[from] serde_json::Error), #[error("invalid at-uri synthesized from frame: {0}")] @@ -106,6 +108,9 @@ pub struct IngestRuntime { pub search: Arc, pub records: Arc, pub resolver: Arc, + pub clock: Arc, + pub entropy: Arc, + pub ws: Arc, pub cancel: CancellationToken, } @@ -117,6 +122,9 @@ impl Clone for IngestRuntime { search: self.search.clone(), records: self.records.clone(), resolver: self.resolver.clone(), + clock: self.clock.clone(), + entropy: self.entropy.clone(), + ws: self.ws.clone(), cancel: self.cancel.clone(), } } @@ -151,7 +159,7 @@ pub async fn run( backoff = RECONNECT_INITIAL_DELAY; } else { tokio::select! { - _ = tokio::time::sleep(jittered(backoff)) => {} + _ = runtime.clock.sleep(jittered(backoff, &*runtime.entropy)) => {} _ = runtime.cancel.cancelled() => return Ok(()), } backoff = (backoff * 2).min(RECONNECT_MAX_DELAY); @@ -167,13 +175,10 @@ fn next_connect_cursor(snapshot: Coverage, start: HydrantCursor) -> HydrantCurso } } -fn jittered(base: Duration) -> Duration { - let entropy = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map(|d| d.subsec_nanos() as u64) - .unwrap_or(0); - let cap_ms = (base.as_millis() as u64 / 4).max(1); - base + Duration::from_millis(entropy % cap_ms) +fn jittered(base: Duration, entropy: &dyn Entropy) -> Duration { + let base_ms = u64::try_from(base.as_millis()).unwrap_or(u64::MAX); + let cap_ms = (base_ms / 4).max(1); + base + Duration::from_millis(entropy.next_u64() % cap_ms) } async fn run_session( @@ -196,18 +201,20 @@ async fn run_session( _ = runtime.cancel.cancelled() => { return SessionEnd { outcome: SessionOutcome::Empty, error: None }; } - res = tokio_tungstenite::connect_async(url.as_str()) => res, + res = runtime.ws.connect(url) => res, }; - let (ws, _resp) = match connect { - Ok(pair) => pair, + let WsConn { + sink: mut ws_sink, + stream: ws_stream, + } = match connect { + Ok(c) => c, Err(e) => { return SessionEnd { outcome: SessionOutcome::Empty, - error: Some(IngestError::Transport(e)), + error: Some(IngestError::Network(e)), }; } }; - let (mut ws_sink, mut ws_stream) = ws.split(); let (frame_tx, mut frame_rx) = tokio::sync::mpsc::channel::(FRAME_CHANNEL_DEPTH); let processor_runtime = runtime.clone(); @@ -218,6 +225,7 @@ async fn run_session( _ = processor_runtime.cancel.cancelled() => break, next = frame_rx.recv() => { let Some(frame) = next else { break }; + let now = processor_runtime.clock.now_unix_micros(); handle_frame( frame, &processor_runtime.store, @@ -225,6 +233,7 @@ async fn run_session( &*processor_runtime.search, &*processor_runtime.records, &processor_runtime.resolver, + now, ) .await; } @@ -235,90 +244,36 @@ async fn run_session( let (control_tx, mut control_rx) = tokio::sync::mpsc::channel::(CONTROL_CHANNEL_DEPTH); let session_cancel = runtime.cancel.child_token(); let reader_cancel = session_cancel.clone(); - let reader = tokio::spawn(async move { - let mut outcome = SessionOutcome::Empty; - let error: Option = loop { - tokio::select! { - biased; - _ = reader_cancel.cancelled() => break None, - msg = ws_stream.next() => { - let Some(msg) = msg else { break None; }; - let parsed = match msg { - Ok(m) => m, - Err(e) => break Some(IngestError::Transport(e)), - }; - match parsed { - Message::Text(text) => { - let frame: HydrantFrame = match serde_json::from_str(&text) { - Ok(f) => f, - Err(e) => break Some(IngestError::Decode(e)), - }; - tokio::select! { - biased; - _ = reader_cancel.cancelled() => break None, - res = frame_tx.send(frame) => { - if res.is_err() { break None; } - } - } - outcome = SessionOutcome::Progressed; - } - Message::Binary(_) => { - debug!("hydrant sent unexpected binary frame, ignoring"); - } - Message::Ping(payload) => { - if control_tx.send(WsEvent::IncomingPing(payload)).await.is_err() { - break None; - } - } - Message::Pong(_) => { - if control_tx.send(WsEvent::IncomingPong).await.is_err() { - break None; - } - } - Message::Frame(_) => {} - Message::Close(close) => { - debug!(?close, "hydrant closed stream"); - break None; - } - } - } - } - }; - SessionEnd { outcome, error } - }); + let reader = tokio::spawn(reader_loop(ws_stream, frame_tx, control_tx, reader_cancel)); - let mut pinger = interval(PING_INTERVAL); - pinger.set_missed_tick_behavior(MissedTickBehavior::Delay); - pinger.tick().await; + let mut next_ping = runtime.clock.now_instant() + PING_INTERVAL; let mut pong_deadline: Option = None; let writer_error: Option = loop { tokio::select! { biased; _ = runtime.cancel.cancelled() => { - let close_frame = CloseFrame { code: CloseCode::Normal, reason: "bobbin shutdown".into() }; - if let Err(e) = ws_sink.send(Message::Close(Some(close_frame))).await { - debug!(?e, "could not send hydrant close frame on shutdown"); - } + let _ = ws_sink.send(WsMessage::Close { code: NORMAL_CLOSE, reason: "bobbin shutdown".to_owned() }).await; break None; } - _ = pinger.tick() => { + _ = runtime.clock.sleep_until(next_ping) => { + next_ping = runtime.clock.now_instant() + PING_INTERVAL; if pong_deadline.is_none() { - if let Err(e) = ws_sink.send(Message::Ping(Bytes::new())).await { - break Some(IngestError::Transport(e)); + if let Err(e) = ws_sink.send(WsMessage::Ping(Bytes::new())).await { + break Some(IngestError::Network(e)); } - pong_deadline = Some(Instant::now() + PONG_TIMEOUT); + pong_deadline = Some(runtime.clock.now_instant() + PONG_TIMEOUT); } } - _ = wait_until(pong_deadline) => { + _ = wait_until(pong_deadline, runtime.clock.as_ref()) => { break Some(IngestError::PongTimeout(PONG_TIMEOUT)); } evt = control_rx.recv() => { let Some(evt) = evt else { break None; }; match evt { WsEvent::IncomingPing(payload) => { - if let Err(e) = ws_sink.send(Message::Pong(payload)).await { - break Some(IngestError::Transport(e)); + if let Err(e) = ws_sink.send(WsMessage::Pong(payload)).await { + break Some(IngestError::Network(e)); } } WsEvent::IncomingPong => { @@ -346,15 +301,71 @@ async fn run_session( } } +async fn reader_loop( + mut ws_stream: Box, + frame_tx: tokio::sync::mpsc::Sender, + control_tx: tokio::sync::mpsc::Sender, + cancel: CancellationToken, +) -> SessionEnd { + let mut outcome = SessionOutcome::Empty; + let error: Option = loop { + tokio::select! { + biased; + _ = cancel.cancelled() => break None, + msg = ws_stream.next() => { + let Some(msg) = msg else { break None; }; + let parsed = match msg { + Ok(m) => m, + Err(e) => break Some(IngestError::Network(e)), + }; + match parsed { + WsMessage::Text(text) => { + let frame: HydrantFrame = match serde_json::from_str(&text) { + Ok(f) => f, + Err(e) => break Some(IngestError::Decode(e)), + }; + tokio::select! { + biased; + _ = cancel.cancelled() => break None, + res = frame_tx.send(frame) => { + if res.is_err() { break None; } + } + } + outcome = SessionOutcome::Progressed; + } + WsMessage::Binary(_) => { + debug!("hydrant sent unexpected binary frame, ignoring"); + } + WsMessage::Ping(payload) => { + if control_tx.send(WsEvent::IncomingPing(payload)).await.is_err() { + break None; + } + } + WsMessage::Pong(_) => { + if control_tx.send(WsEvent::IncomingPong).await.is_err() { + break None; + } + } + WsMessage::Close { code, reason } => { + debug!(code, %reason, "hydrant closed stream"); + break None; + } + } + } + } + }; + SessionEnd { outcome, error } +} + #[derive(Debug)] enum WsEvent { IncomingPing(Bytes), IncomingPong, } -async fn wait_until(deadline: Option) { +async fn wait_until(deadline: Option, clock: &dyn Clock) { match deadline { - Some(d) => tokio::time::sleep_until(d).await, + Some(d) => clock.sleep_until(d).await, None => std::future::pending::<()>().await, } } @@ -366,9 +377,10 @@ async fn handle_frame( search: &S, records: &dyn RecordStore, resolver: &RepoIdResolver, + now: UnixMicros, ) { let cursor = HydrantCursor::new(frame.id); - let signal = promotion_signal(frame.record.as_ref(), now_micros()); + let signal = promotion_signal(frame.record.as_ref(), now); coverage.update(|c| c.advance(cursor).maybe_promote(signal)); match frame.kind { @@ -394,22 +406,15 @@ async fn handle_frame( } } -fn promotion_signal(record: Option<&RecordFrame>, now_micros: u64) -> PromotionSignal { +fn promotion_signal(record: Option<&RecordFrame>, now: UnixMicros) -> PromotionSignal { PromotionSignal { live: record.is_some_and(|r| r.live), rev_micros: record.map(|r| r.rev.timestamp()), - now_micros, + now_micros: now.raw(), skew_micros: READY_SKEW.as_micros() as u64, } } -fn now_micros() -> u64 { - SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("system clock before unix epoch") - .as_micros() as u64 -} - async fn apply_record( record: RecordFrame, store: &EdgeStore, @@ -538,6 +543,7 @@ mod tests { use super::*; use bobbin_edge_index::Coverage; use bobbin_record_lru::{CacheCapacity, LruRecordStore, NoopRecordStore, RecordStore}; + use bobbin_runtime::{OsEntropy, RuntimeHasher, SystemClock, TungsteniteWs}; use bobbin_types::search::NoopSearchSink; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::tid::Tid; @@ -547,12 +553,16 @@ mod tests { fn fresh() -> (Arc, Arc, Arc) { ( - Arc::new(EdgeStore::new()), + Arc::new(EdgeStore::new(RuntimeHasher::default())), Arc::new(CoverageWatch::new()), - Arc::new(RepoIdResolver::detached()), + Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), ) } + fn now() -> UnixMicros { + SystemClock::new().now_unix_micros() + } + fn parse_frame(value: serde_json::Value) -> HydrantFrame { let text = serde_json::to_string(&value).expect("serialize fixture"); serde_json::from_str(&text).expect("deserialize fixture") @@ -585,6 +595,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert_eq!(store.key_count(), 0); @@ -618,6 +629,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( @@ -646,6 +658,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert_eq!(store.count(&key), 0); @@ -680,6 +693,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; handle_frame( @@ -689,6 +703,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -727,7 +742,7 @@ mod tests { })); let source = AtUri::new_owned("at://did:plc:olaren/sh.tangled.feed.star/abcabcabcabcz").unwrap(); - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; let cached = lru.get(&source).expect("hydrant cid must seed the lru"); assert_eq!(cached.cid.as_ref(), VALID_CID); let parsed: serde_json::Value = serde_json::from_slice(&cached.value).unwrap(); @@ -766,7 +781,7 @@ mod tests { } } })); - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; assert!( lru.get(&source).is_none(), "missing cid means we cannot trust the body, so the lru must be cleared", @@ -801,7 +816,7 @@ mod tests { "record": null } })); - handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver).await; + handle_frame(frame, &store, &cov, &NoopSearchSink, &lru, &resolver, now()).await; assert!(lru.get(&source).is_none()); } @@ -833,6 +848,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert!(cov.snapshot().is_ready()); @@ -867,6 +883,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert!(!cov.snapshot().is_ready()); @@ -935,6 +952,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert_eq!(store.key_count(), 0); @@ -957,6 +975,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert_eq!(store.key_count(), 0); @@ -975,6 +994,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; assert_eq!(cov.snapshot().last_cursor(), HydrantCursor::new(8)); @@ -982,11 +1002,14 @@ mod tests { fn fresh_runtime(cancel: CancellationToken) -> IngestRuntime { IngestRuntime { - store: Arc::new(EdgeStore::new()), + store: Arc::new(EdgeStore::new(RuntimeHasher::default())), coverage: Arc::new(CoverageWatch::new()), search: Arc::new(NoopSearchSink), records: Arc::new(NoopRecordStore) as Arc, - resolver: Arc::new(RepoIdResolver::detached()), + resolver: Arc::new(RepoIdResolver::detached(RuntimeHasher::default())), + clock: Arc::new(SystemClock::new()), + entropy: Arc::new(OsEntropy), + ws: TungsteniteWs::shared(), cancel, } } @@ -1009,8 +1032,9 @@ mod tests { fn jittered_stays_within_one_quarter_of_base() { let base = Duration::from_secs(1); let cap = base + Duration::from_millis(250); + let entropy = OsEntropy; (0..50).for_each(|_| { - let j = jittered(base); + let j = jittered(base, &entropy); assert!(j >= base, "jitter must not undershoot"); assert!(j <= cap, "jitter must not exceed +25%, got {:?}", j); }); @@ -1045,6 +1069,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -1072,6 +1097,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -1121,6 +1147,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -1173,6 +1200,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -1200,6 +1228,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; @@ -1241,6 +1270,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( @@ -1285,6 +1315,7 @@ mod tests { &NoopSearchSink, &NoopRecordStore, &resolver, + now(), ) .await; let key = bobbin_types::ids::EdgeKey::new( diff --git a/crates/ingest/src/resolver.rs b/crates/ingest/src/resolver.rs index 9eeaf22..3836d92 100644 --- a/crates/ingest/src/resolver.rs +++ b/crates/ingest/src/resolver.rs @@ -1,3 +1,4 @@ +use bobbin_runtime::RuntimeHasher; use bobbin_slingshot_client::{SlingshotClient, SlingshotError}; use bobbin_types::edges::{ExtractError, Record}; use bobbin_types::ids::nsid_static; @@ -61,21 +62,21 @@ impl CacheEntry { } pub struct RepoIdResolver { - cache: SccMap, + cache: SccMap, client: Option, } impl RepoIdResolver { - pub fn with_slingshot(client: SlingshotClient) -> Self { + pub fn with_slingshot(client: SlingshotClient, hasher: RuntimeHasher) -> Self { Self { - cache: SccMap::new(), + cache: SccMap::with_hasher(hasher), client: Some(client), } } - pub fn detached() -> Self { + pub fn detached(hasher: RuntimeHasher) -> Self { Self { - cache: SccMap::new(), + cache: SccMap::with_hasher(hasher), client: None, } } @@ -197,7 +198,7 @@ mod tests { #[tokio::test] async fn observation_with_repo_did_resolves_mapped() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), @@ -213,7 +214,7 @@ mod tests { #[tokio::test] async fn observation_without_repo_did_resolves_no_repo_did() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe(did("did:plc:nel"), rkey("abcabcabcabcz"), None) .await; @@ -229,7 +230,7 @@ mod tests { #[tokio::test] async fn cache_miss_without_client_is_unresolvable() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let got = resolver .resolve(&did("did:plc:nel"), &rkey("abcabcabcabcz")) .await; @@ -238,7 +239,7 @@ mod tests { #[tokio::test] async fn observation_overwrites_prior_value() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); resolver .observe( did("did:plc:nel"), @@ -261,7 +262,7 @@ mod tests { #[tokio::test] async fn fill_provisional_does_not_downgrade_authoritative_mapped() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver @@ -283,7 +284,7 @@ mod tests { #[tokio::test] async fn fill_provisional_does_not_downgrade_authoritative_no_repo_did() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver.observe(owner.clone(), key.clone(), None).await; @@ -311,8 +312,9 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = RepoIdResolver::with_slingshot(client); + let client = + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); @@ -336,8 +338,9 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = RepoIdResolver::with_slingshot(client); + let client = + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); @@ -366,8 +369,9 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = RepoIdResolver::with_slingshot(client); + let client = + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); @@ -392,8 +396,9 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = RepoIdResolver::with_slingshot(client); + let client = + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); @@ -417,8 +422,9 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(url::Url::parse(&server.uri()).unwrap()).unwrap(); - let resolver = RepoIdResolver::with_slingshot(client); + let client = + SlingshotClient::with_default_http(url::Url::parse(&server.uri()).unwrap()).unwrap(); + let resolver = RepoIdResolver::with_slingshot(client, RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); @@ -430,7 +436,7 @@ mod tests { #[tokio::test] async fn firehose_observe_can_demote_provisional() { - let resolver = RepoIdResolver::detached(); + let resolver = RepoIdResolver::detached(RuntimeHasher::default()); let owner = did("did:plc:nel"); let key = rkey("abcabcabcabcz"); resolver diff --git a/crates/knot-proxy/Cargo.toml b/crates/knot-proxy/Cargo.toml index 027aa93..c946974 100644 --- a/crates/knot-proxy/Cargo.toml +++ b/crates/knot-proxy/Cargo.toml @@ -6,6 +6,7 @@ license.workspace = true rust-version.workspace = true [dependencies] +bobbin-runtime = { workspace = true } bytes = { workspace = true } futures = { workspace = true } http = { workspace = true } diff --git a/crates/knot-proxy/src/breaker.rs b/crates/knot-proxy/src/breaker.rs index 1416d29..881f569 100644 --- a/crates/knot-proxy/src/breaker.rs +++ b/crates/knot-proxy/src/breaker.rs @@ -1,7 +1,9 @@ use std::sync::{Arc, Mutex}; -use std::time::{Duration, Instant}; +use std::time::Duration; +use bobbin_runtime::Clock; use thiserror::Error; +use tokio::time::Instant; #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub struct FailureThreshold(u32); @@ -34,24 +36,35 @@ enum BreakerState { #[error("circuit breaker open")] pub struct CircuitOpen; -#[derive(Debug)] pub struct Breaker { state: Mutex, threshold: FailureThreshold, cooldown: Duration, + clock: Arc, +} + +impl std::fmt::Debug for Breaker { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Breaker") + .field("state", &self.state) + .field("threshold", &self.threshold) + .field("cooldown", &self.cooldown) + .finish_non_exhaustive() + } } impl Breaker { - pub fn new(threshold: FailureThreshold, cooldown: Duration) -> Self { + pub fn new(threshold: FailureThreshold, cooldown: Duration, clock: Arc) -> Self { Self { state: Mutex::new(BreakerState::Closed { failures: 0 }), threshold, cooldown, + clock, } } pub fn try_acquire(self: &Arc) -> Result { - self.try_acquire_at(Instant::now()) + self.try_acquire_at(self.clock.now_instant()) } fn try_acquire_at(self: &Arc, now: Instant) -> Result { @@ -72,7 +85,7 @@ impl Breaker { } pub fn record_failure(&self) { - self.record_failure_at(Instant::now()); + self.record_failure_at(self.clock.now_instant()); } fn record_failure_at(&self, now: Instant) { @@ -134,11 +147,13 @@ impl Drop for BreakerPermit { #[cfg(test)] mod tests { use super::*; + use bobbin_runtime::SystemClock; fn breaker(threshold: u32, cooldown_ms: u64) -> Arc { Arc::new(Breaker::new( FailureThreshold::new(threshold).unwrap(), Duration::from_millis(cooldown_ms), + Arc::new(SystemClock::new()), )) } diff --git a/crates/knot-proxy/src/lib.rs b/crates/knot-proxy/src/lib.rs index da29312..097a653 100644 --- a/crates/knot-proxy/src/lib.rs +++ b/crates/knot-proxy/src/lib.rs @@ -3,10 +3,14 @@ use std::sync::Arc; use std::task::{Context, Poll}; use std::time::Duration; +use bobbin_runtime::{ + BodyStream as InnerBodyStream, Clock, HttpRequest, HttpResponseHead, HttpTransport, + NetworkError, ReqwestHttp, RuntimeHasher, +}; use bytes::Bytes; use futures::Stream; -use http::HeaderMap; -use reqwest::{Client, Response, StatusCode, redirect::Policy}; +use http::{HeaderMap, StatusCode}; +use reqwest::{Client, redirect::Policy}; use scc::HashMap as SccMap; use thiserror::Error; use url::Url; @@ -25,8 +29,6 @@ const HTTPS_SCHEME: &str = "https"; pub struct KnotProxyConfig { pub failure_threshold: FailureThreshold, pub cooldown: Duration, - pub connect_timeout: Duration, - pub read_timeout: Duration, pub allow_private_hosts: bool, pub require_https: bool, } @@ -36,14 +38,27 @@ impl Default for KnotProxyConfig { Self { failure_threshold: FailureThreshold::new(5).expect("nonzero literal"), cooldown: Duration::from_secs(30), - connect_timeout: Duration::from_secs(5), - read_timeout: Duration::from_secs(60), allow_private_hosts: false, require_https: true, } } } +#[derive(Clone, Copy, Debug)] +pub struct KnotHttpConfig { + pub connect_timeout: Duration, + pub read_timeout: Duration, +} + +impl Default for KnotHttpConfig { + fn default() -> Self { + Self { + connect_timeout: Duration::from_secs(5), + read_timeout: Duration::from_secs(60), + } + } +} + #[derive(Debug, Error)] pub enum KnotProxyError { #[error("circuit breaker open")] @@ -68,35 +83,56 @@ pub enum KnotProxyError { } pub struct KnotProxy { - http: Client, - breakers: SccMap>, + http: Arc, + breakers: SccMap, RuntimeHasher>, threshold: FailureThreshold, cooldown: Duration, allow_private_hosts: bool, require_https: bool, + clock: Arc, } impl KnotProxy { - pub fn new(config: KnotProxyConfig) -> Result { + pub fn new( + config: KnotProxyConfig, + http: KnotHttpConfig, + clock: Arc, + hasher: RuntimeHasher, + ) -> Result { let resolver = Arc::new(dns::PrivateAddressFilter::new(config.allow_private_hosts)); - let http = Client::builder() + let client = Client::builder() .user_agent(USER_AGENT) - .connect_timeout(config.connect_timeout) - .read_timeout(config.read_timeout) + .connect_timeout(http.connect_timeout) + .read_timeout(http.read_timeout) .redirect(Policy::none()) .no_gzip() .no_brotli() .no_deflate() .dns_resolver(resolver) .build()?; - Ok(Self { + Ok(Self::with_transport( + ReqwestHttp::shared(client), + config, + clock, + hasher, + )) + } + + pub fn with_transport( + http: Arc, + config: KnotProxyConfig, + clock: Arc, + hasher: RuntimeHasher, + ) -> Self { + Self { http, - breakers: SccMap::new(), + breakers: SccMap::with_hasher(hasher), threshold: config.failure_threshold, cooldown: config.cooldown, allow_private_hosts: config.allow_private_hosts, require_https: config.require_https, - }) + clock, + } } pub fn allows_private_hosts(&self) -> bool { @@ -120,7 +156,7 @@ impl KnotProxy { .try_acquire() .map_err(|_: CircuitOpen| KnotProxyError::CircuitOpen)?; let url = build_xrpc_url(host, nsid, query); - let outcome = self.http.get(url).headers(headers).send().await; + let outcome = self.http.execute(HttpRequest { url, headers }).await; classify(outcome, permit) } @@ -148,43 +184,57 @@ impl KnotProxy { let entry = self.breakers.entry_async(host.clone()).await; Arc::clone( entry - .or_insert_with(|| Arc::new(Breaker::new(self.threshold, self.cooldown))) + .or_insert_with(|| { + Arc::new(Breaker::new( + self.threshold, + self.cooldown, + self.clock.clone(), + )) + }) .get(), ) } } -#[derive(Debug)] pub struct ProxyResponse { - inner: Response, + status: StatusCode, + headers: HeaderMap, + body: InnerBodyStream, permit: BreakerPermit, } +impl std::fmt::Debug for ProxyResponse { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("ProxyResponse") + .field("status", &self.status) + .field("headers", &self.headers) + .finish_non_exhaustive() + } +} + impl ProxyResponse { pub fn status(&self) -> StatusCode { - self.inner.status() + self.status } pub fn headers(&self) -> &HeaderMap { - self.inner.headers() + &self.headers } pub fn into_body_stream(self) -> BodyStream { - BodyStream::new(self.inner, self.permit) + BodyStream::new(self.body, self.permit) } } -type ChunkStream = Pin> + Send>>; - pub struct BodyStream { - inner: ChunkStream, + inner: InnerBodyStream, permit: Option, } impl BodyStream { - fn new(response: Response, permit: BreakerPermit) -> Self { + fn new(inner: InnerBodyStream, permit: BreakerPermit) -> Self { Self { - inner: Box::pin(response.bytes_stream()), + inner, permit: Some(permit), } } @@ -201,7 +251,7 @@ impl BodyStream { } impl Stream for BodyStream { - type Item = Result; + type Item = Result; fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { let next = match self.inner.as_mut().poll_next(cx) { @@ -229,22 +279,24 @@ fn build_xrpc_url(host: &KnotHost, nsid: &str, query: &[(&str, &str)]) -> Url { } fn classify( - outcome: Result, + outcome: Result, permit: BreakerPermit, ) -> Result { match outcome { - Ok(resp) if is_upstream_failure(resp.status()) => { - let status = resp.status(); + Ok(head) if is_upstream_failure(head.status) => { + let status = head.status; permit.record_failure(); Err(KnotProxyError::Upstream(status)) } - Ok(resp) => Ok(ProxyResponse { - inner: resp, + Ok(head) => Ok(ProxyResponse { + status: head.status, + headers: head.headers, + body: head.body, permit, }), Err(err) => { permit.record_failure(); - Err(map_transport(err)) + Err(map_network(err)) } } } @@ -257,22 +309,21 @@ fn is_unfollowable_redirect(status: StatusCode) -> bool { matches!(status.as_u16(), 301 | 302 | 303 | 307 | 308) } -fn map_transport(err: reqwest::Error) -> KnotProxyError { - let msg = err.to_string(); - if err.is_timeout() { - KnotProxyError::Timeout(msg) - } else if err.is_connect() { - KnotProxyError::Connect(msg) - } else if err.is_redirect() { - KnotProxyError::Redirect(msg) - } else { - KnotProxyError::Transport(msg) +fn map_network(err: NetworkError) -> KnotProxyError { + match err { + NetworkError::Timeout(msg) => KnotProxyError::Timeout(msg), + NetworkError::Connect(msg) => KnotProxyError::Connect(msg), + NetworkError::Redirect(msg) => KnotProxyError::Redirect(msg), + NetworkError::Transport(msg) | NetworkError::Body(msg) | NetworkError::Protocol(msg) => { + KnotProxyError::Transport(msg) + } } } #[cfg(test)] mod tests { use super::*; + use bobbin_runtime::SystemClock; use futures::stream::TryStreamExt; use tokio::io::AsyncWriteExt; use wiremock::matchers::{method, path, query_param}; @@ -282,13 +333,28 @@ mod tests { KnotProxyConfig { failure_threshold: FailureThreshold::new(2).unwrap(), cooldown: Duration::from_millis(80), - connect_timeout: Duration::from_millis(500), - read_timeout: Duration::from_secs(2), allow_private_hosts: true, require_https: false, } } + pub(crate) fn http_config_for_test() -> KnotHttpConfig { + KnotHttpConfig { + connect_timeout: Duration::from_millis(500), + read_timeout: Duration::from_secs(2), + } + } + + fn proxy_for_test() -> KnotProxy { + KnotProxy::new( + config_for_test(), + http_config_for_test(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap() + } + async fn server() -> MockServer { MockServer::start().await } @@ -297,7 +363,7 @@ mod tests { KnotHost::parse(&server.uri()).unwrap() } - pub(crate) async fn drain(stream: BodyStream) -> Result { + pub(crate) async fn drain(stream: BodyStream) -> Result { let chunks: Vec = stream.try_collect().await?; let total: usize = chunks.iter().map(|b| b.len()).sum(); let mut buf = bytes::BytesMut::with_capacity(total); @@ -321,7 +387,7 @@ mod tests { .mount(&server) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let resp = proxy .forward( &host_of(&server), @@ -349,7 +415,7 @@ mod tests { .mount(&server) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let host = host_of(&server); let r1 = proxy .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) @@ -374,7 +440,7 @@ mod tests { .mount(&server) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let host = host_of(&server); let r1 = proxy .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) @@ -413,7 +479,7 @@ mod tests { .mount(&server) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let host = host_of(&server); let _ = proxy .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) @@ -456,7 +522,7 @@ mod tests { .mount(&good) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let bad_host = host_of(&bad); let good_host = host_of(&good); let _ = proxy @@ -499,7 +565,13 @@ mod tests { allow_private_hosts: false, ..config_for_test() }; - let proxy = KnotProxy::new(strict).unwrap(); + let proxy = KnotProxy::new( + strict, + http_config_for_test(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(); let err = proxy .forward( &host_of(&server), @@ -519,7 +591,13 @@ mod tests { require_https: true, ..config_for_test() }; - let proxy = KnotProxy::new(strict).unwrap(); + let proxy = KnotProxy::new( + strict, + http_config_for_test(), + Arc::new(SystemClock::new()), + RuntimeHasher::default(), + ) + .unwrap(); let err = proxy .forward( &host_of(&server), @@ -542,7 +620,7 @@ mod tests { drop(listener); let dead = KnotHost::parse(&format!("http://{addr}")).unwrap(); - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let r1 = proxy .forward(&dead, "sh.tangled.repo.blob", &[], HeaderMap::new()) .await; @@ -581,7 +659,7 @@ mod tests { .mount(&secondary) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let err = proxy .forward( &host_of(&primary), @@ -607,7 +685,7 @@ mod tests { .respond_with(ResponseTemplate::new(304).insert_header("etag", "\"v1\"")) .mount(&server) .await; - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let resp = proxy .forward( &host_of(&server), @@ -643,7 +721,7 @@ mod tests { }); let host = KnotHost::parse(&format!("http://{addr}")).unwrap(); - let proxy = KnotProxy::new(config_for_test()).unwrap(); + let proxy = proxy_for_test(); let r1 = proxy .forward(&host, "sh.tangled.repo.blob", &[], HeaderMap::new()) diff --git a/crates/slingshot-client/Cargo.toml b/crates/slingshot-client/Cargo.toml index 9fc1961..06af319 100644 --- a/crates/slingshot-client/Cargo.toml +++ b/crates/slingshot-client/Cargo.toml @@ -6,12 +6,14 @@ license.workspace = true rust-version.workspace = true [dependencies] +bobbin-runtime = { workspace = true } bobbin-types = { workspace = true } jacquard-common = { workspace = true } bytes = { workspace = true } cid = { workspace = true } futures = { workspace = true } +http = { workspace = true } reqwest = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } diff --git a/crates/slingshot-client/src/lib.rs b/crates/slingshot-client/src/lib.rs index ba39279..ece71bd 100644 --- a/crates/slingshot-client/src/lib.rs +++ b/crates/slingshot-client/src/lib.rs @@ -1,17 +1,18 @@ use std::sync::Arc; use std::time::Duration; +use bobbin_runtime::{HttpRequest, HttpResponseHead, HttpTransport, NetworkError, ReqwestHttp}; use bobbin_types::record::RecordBody; use bytes::{Bytes, BytesMut}; use cid::Cid as IpldCid; use futures::TryStreamExt; +use http::{HeaderMap, StatusCode}; use jacquard_common::BosStr; use jacquard_common::types::did::Did; use jacquard_common::types::ident::AtIdentifier; use jacquard_common::types::nsid::Nsid; use jacquard_common::types::recordkey::Rkey; use jacquard_common::types::string::{AtStrError, AtUri}; -use reqwest::{Client, Response, StatusCode}; use serde::Deserialize; use serde_json::value::RawValue; use thiserror::Error; @@ -24,18 +25,28 @@ const GET_RECORD_PATH: &str = "xrpc/com.atproto.repo.getRecord"; const RESOLVE_MINI_DOC_PATH: &str = "xrpc/com.bad-example.identity.resolveMiniDoc"; pub const MAX_BODY_BYTES: u64 = 4 * 1024 * 1024; -#[derive(Clone, Debug)] +#[derive(Clone)] pub struct SlingshotClient { - http: Client, + http: Arc, base: Url, } +impl std::fmt::Debug for SlingshotClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SlingshotClient") + .field("base", &self.base) + .finish_non_exhaustive() + } +} + #[derive(Debug, Error)] pub enum SlingshotError { #[error("invalid base url scheme: {0}")] BadScheme(String), - #[error("transport: {0}")] - Transport(#[from] reqwest::Error), + #[error("network: {0}")] + Network(#[from] NetworkError), + #[error("http client build: {0}")] + Build(String), #[error("record not found")] NotFound, #[error("upstream returned status {0}")] @@ -54,21 +65,29 @@ pub enum SlingshotError { UriMismatch { expected: String, got: String }, } +pub fn default_http_client() -> Result { + reqwest::Client::builder() + .user_agent(USER_AGENT) + .timeout(REQUEST_TIMEOUT) + .connect_timeout(CONNECT_TIMEOUT) + .build() +} + impl SlingshotClient { - pub fn new(base: Url) -> Result { + pub fn new(base: Url, http: Arc) -> Result { match base.scheme() { "http" | "https" => {} other => return Err(SlingshotError::BadScheme(other.to_owned())), } let base = ensure_trailing_slash(base); - let http = Client::builder() - .user_agent(USER_AGENT) - .timeout(REQUEST_TIMEOUT) - .connect_timeout(CONNECT_TIMEOUT) - .build()?; Ok(Self { http, base }) } + pub fn with_default_http(base: Url) -> Result { + let client = default_http_client().map_err(|e| SlingshotError::Build(e.to_string()))?; + Self::new(base, ReqwestHttp::shared(client)) + } + pub async fn resolve_mini_doc( &self, identifier: &AtIdentifier, @@ -83,8 +102,14 @@ impl SlingshotClient { .clear() .append_pair("identifier", identifier.as_str()); - let resp = self.http.get(url).send().await?; - match resp.status() { + let resp = self + .http + .execute(HttpRequest { + url, + headers: HeaderMap::new(), + }) + .await?; + match resp.status { StatusCode::OK => read_bounded(resp).await, StatusCode::NOT_FOUND => Err(SlingshotError::NotFound), other => Err(SlingshotError::Upstream(other)), @@ -110,8 +135,14 @@ impl SlingshotClient { .append_pair("collection", collection.as_ref()) .append_pair("rkey", rkey.as_ref()); - let resp = self.http.get(url).send().await?; - match resp.status() { + let resp = self + .http + .execute(HttpRequest { + url, + headers: HeaderMap::new(), + }) + .await?; + match resp.status { StatusCode::OK => { let bytes = read_bounded(resp).await?; let body = decode(&bytes)?; @@ -132,18 +163,15 @@ fn ensure_trailing_slash(mut base: Url) -> Url { base } -async fn read_bounded(resp: Response) -> Result { - if resp - .content_length() - .is_some_and(|len| len > MAX_BODY_BYTES) - { +async fn read_bounded(resp: HttpResponseHead) -> Result { + if resp.content_length.is_some_and(|len| len > MAX_BODY_BYTES) { return Err(SlingshotError::BodyTooLarge { limit: MAX_BODY_BYTES, }); } let buf = resp - .bytes_stream() - .map_err(SlingshotError::Transport) + .body + .map_err(SlingshotError::Network) .try_fold(BytesMut::new(), |mut acc, chunk| async move { if (acc.len() as u64).saturating_add(chunk.len() as u64) > MAX_BODY_BYTES { return Err(SlingshotError::BodyTooLarge { @@ -217,6 +245,10 @@ mod tests { MockServer::start().await } + fn client_for(server: &MockServer) -> SlingshotClient { + SlingshotClient::with_default_http(Url::parse(&server.uri()).unwrap()).unwrap() + } + #[tokio::test] async fn returns_decoded_record_on_200() { let server = server().await; @@ -238,7 +270,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let resp = client .get_record( &did("did:plc:abalone"), @@ -268,7 +300,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let resp = client .get_record(&did("did:plc:uni"), &nsid("sh.tangled.repo"), &rkey("r1")) .await @@ -285,7 +317,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -306,7 +338,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -330,7 +362,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -359,7 +391,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -388,7 +420,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -417,7 +449,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -449,7 +481,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -477,7 +509,7 @@ mod tests { .mount(&server) .await; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let err = client .get_record( &did("did:plc:abalone"), @@ -494,7 +526,8 @@ mod tests { #[test] fn rejects_non_http_scheme() { - let err = SlingshotClient::new(Url::parse("ftp://nel.pet").unwrap()).expect_err("ftp bad"); + let err = SlingshotClient::with_default_http(Url::parse("ftp://nel.pet").unwrap()) + .expect_err("ftp bad"); assert!(matches!(err, SlingshotError::BadScheme(s) if s == "ftp")); } @@ -518,7 +551,7 @@ mod tests { .await; let base = Url::parse(&format!("{}/api/v0", server.uri())).unwrap(); - let client = SlingshotClient::new(base).unwrap(); + let client = SlingshotClient::with_default_http(base).unwrap(); client .get_record( &did("did:plc:abalone"), @@ -555,7 +588,7 @@ mod tests { jacquard_common::types::ident::AtIdentifier::Handle(_) => unreachable!(), }; - let client = SlingshotClient::new(Url::parse(&server.uri()).unwrap()).unwrap(); + let client = client_for(&server); let resp = client .get_record(&did_borrow, &collection, &rkey) .await -- 2.51.2