+78 -3

crates/ingest/src/lib.rs

Reviewed

··· 190 190 backoff = RECONNECT_INITIAL_DELAY; 191 191 } else { 192 192 tokio::select! { 193 193 + biased; 194 194 + _ = runtime.cancel.cancelled() => return Ok(()), 193 195 _ = runtime.clock.sleep(jittered(backoff, &*runtime.entropy)) => {} 194 194 - _ = runtime.cancel.cancelled() => return Ok(()), 195 196 } 196 197 backoff = (backoff * 2).min(RECONNECT_MAX_DELAY); 197 198 } ··· 518 519 } 519 520 } 520 521 522 522 + #[derive(Clone, Copy, Debug, Eq, PartialEq)] 523 523 + enum Regime { 524 524 + Replay, 525 525 + Live, 526 526 + NonRecord, 527 527 + } 528 528 + 529 529 + impl Regime { 530 530 + fn as_str(self) -> &'static str { 531 531 + match self { 532 532 + Self::Replay => "replay", 533 533 + Self::Live => "live", 534 534 + Self::NonRecord => "non_record", 535 535 + } 536 536 + } 537 537 + } 538 538 + 521 539 struct Pending { 522 540 cursor: HydrantCursor, 523 541 signal: PromotionSignal, 542 542 + regime: Regime, 524 543 op: PendingOp, 525 544 } 526 545 ··· 610 629 ) { 611 630 let nsid = pending_nsid(&staged.pending.op).cloned(); 612 631 let edge_count = pending_edge_count(&staged.pending.op); 632 632 + let regime = staged.pending.regime; 633 633 + let cursor = staged.pending.cursor.raw(); 613 634 let Resolved { 614 635 pending, 615 636 prepare_start, ··· 622 643 let commit_end = rt.clock.now_instant(); 623 644 tracing::trace!( 624 645 target: "bobbin_ingest::stage", 646 646 + cursor, 647 647 + regime = regime.as_str(), 625 648 nsid = %nsid.as_ref().map(Nsid::as_str).unwrap_or(""), 626 649 edge_count, 627 650 prepare_us = prepare_end.duration_since(prepare_start).as_micros() as u64, ··· 642 665 ) -> Pending { 643 666 let cursor = HydrantCursor::new(frame.id); 644 667 let signal = promotion_signal(frame.record.as_ref(), now); 668 668 + let regime = match frame.record.as_ref() { 669 669 + Some(r) if r.live => Regime::Live, 670 670 + Some(_) => Regime::Replay, 671 671 + None => Regime::NonRecord, 672 672 + }; 645 673 let op = match frame.kind { 646 674 FrameKind::Record => prepare_record(frame.record, resolver).await, 647 675 FrameKind::Identity | FrameKind::Account => PendingOp::Noop, ··· 653 681 Pending { 654 682 cursor, 655 683 signal, 684 684 + regime, 656 685 op, 657 686 } 658 687 } ··· 725 754 } 726 755 727 756 async fn resolve_pending(pending: Pending, resolver: &RepoIdResolver) -> Pending { 728 728 - let Pending { cursor, signal, op } = pending; 757 757 + let Pending { 758 758 + cursor, 759 759 + signal, 760 760 + regime, 761 761 + op, 762 762 + } = pending; 729 763 let op = match op { 730 764 PendingOp::Upsert { 731 765 source, ··· 750 784 Pending { 751 785 cursor, 752 786 signal, 787 787 + regime, 753 788 op, 754 789 } 755 790 } ··· 761 796 search: &S, 762 797 records: &dyn RecordStore, 763 798 ) { 764 764 - let Pending { cursor, signal, op } = pending; 799 799 + let Pending { 800 800 + cursor, 801 801 + signal, 802 802 + regime: _, 803 803 + op, 804 804 + } = pending; 765 805 match op { 766 806 PendingOp::Noop => {} 767 807 PendingOp::ClearCache { source } => records.remove(&source), ··· 1065 1105 bobbin_types::ids::EdgeKey::new(kind, AtUri::new_owned("at://did:plc:uni").unwrap()); 1066 1106 assert_eq!(store.count(&old), 0); 1067 1107 assert_eq!(store.count(&new), 1); 1108 1108 + } 1109 1109 + 1110 1110 + #[tokio::test] 1111 1111 + async fn prepare_frame_tags_regime_from_live_flag() { 1112 1112 + let (_store, _cov, resolver) = fresh(); 1113 1113 + let mk = |live: bool| -> HydrantFrame { 1114 1114 + parse_frame(json!({ 1115 1115 + "id": 1, 1116 1116 + "type": "record", 1117 1117 + "record": { 1118 1118 + "live": live, 1119 1119 + "did": "did:plc:olaren", 1120 1120 + "rev": fresh_tid().as_str(), 1121 1121 + "collection": "sh.tangled.feed.star", 1122 1122 + "rkey": "abcabcabcabcz", 1123 1123 + "action": "create", 1124 1124 + "record": { 1125 1125 + "$type": "sh.tangled.feed.star", 1126 1126 + "createdAt": "2026-05-01T00:00:00Z", 1127 1127 + "subjectDid": "did:plc:abalone" 1128 1128 + } 1129 1129 + } 1130 1130 + })) 1131 1131 + }; 1132 1132 + let live_pending = prepare_frame(mk(true), &resolver, now()).await; 1133 1133 + assert_eq!(live_pending.regime, Regime::Live); 1134 1134 + let replay_pending = prepare_frame(mk(false), &resolver, now()).await; 1135 1135 + assert_eq!(replay_pending.regime, Regime::Replay); 1136 1136 + 1137 1137 + let identity: HydrantFrame = parse_frame(json!({ 1138 1138 + "id": 9, 1139 1139 + "type": "identity", 1140 1140 + })); 1141 1141 + let id_pending = prepare_frame(identity, &resolver, now()).await; 1142 1142 + assert_eq!(id_pending.regime, Regime::NonRecord); 1068 1143 } 1069 1144 1070 1145 #[tokio::test]

+69

crates/runtime/src/clock.rs

Reviewed

··· 73 73 } 74 74 } 75 75 76 76 + #[derive(Clone, Debug)] 77 77 + pub struct SimClock { 78 78 + base_unix: UnixMicros, 79 79 + base_instant: Instant, 80 80 + } 81 81 + 82 82 + impl SimClock { 83 83 + pub fn at(base_unix: UnixMicros) -> Self { 84 84 + Self { 85 85 + base_unix, 86 86 + base_instant: Instant::now(), 87 87 + } 88 88 + } 89 89 + } 90 90 + 91 91 + impl Clock for SimClock { 92 92 + fn now_unix_micros(&self) -> UnixMicros { 93 93 + let elapsed = self 94 94 + .now_instant() 95 95 + .saturating_duration_since(self.base_instant); 96 96 + let micros = u64::try_from(elapsed.as_micros()).unwrap_or(u64::MAX); 97 97 + UnixMicros::new(self.base_unix.raw().saturating_add(micros)) 98 98 + } 99 99 + 100 100 + fn now_instant(&self) -> Instant { 101 101 + Instant::now() 102 102 + } 103 103 + 104 104 + fn sleep(&self, duration: Duration) -> SleepFuture { 105 105 + Box::pin(tokio::time::sleep(duration)) 106 106 + } 107 107 + 108 108 + fn sleep_until(&self, deadline: Instant) -> SleepFuture { 109 109 + Box::pin(tokio::time::sleep_until(deadline)) 110 110 + } 111 111 + } 112 112 + 76 113 #[cfg(test)] 77 114 mod tests { 78 115 use super::*; ··· 139 176 assert_eq!( 140 177 dyn_clock.now_instant(), 141 178 instant_before + Duration::from_micros(500), 179 179 + ); 180 180 + } 181 181 + 182 182 + #[tokio::test(start_paused = true)] 183 183 + async fn sim_clock_anchors_unix_at_explicit_base_under_paused_time() { 184 184 + let base = UnixMicros::new(1_700_000_000_000_000); 185 185 + let clock = SimClock::at(base); 186 186 + assert_eq!( 187 187 + clock.now_unix_micros(), 188 188 + base, 189 189 + "sim clock unix axis must reflect the explicit base, not wall clock", 190 190 + ); 191 191 + let bump = Duration::from_secs(10); 192 192 + tokio::time::advance(bump).await; 193 193 + assert_eq!( 194 194 + clock.now_unix_micros().raw(), 195 195 + base.raw() + u64::try_from(bump.as_micros()).unwrap(), 196 196 + "sim clock unix axis must follow tokio virtual advance", 197 197 + ); 198 198 + } 199 199 + 200 200 + #[tokio::test(start_paused = true)] 201 201 + async fn sim_clock_two_constructions_with_same_base_align() { 202 202 + let base = UnixMicros::new(2_000_000_000_000_000); 203 203 + let a = SimClock::at(base); 204 204 + let b = SimClock::at(base); 205 205 + assert_eq!(a.now_unix_micros(), b.now_unix_micros()); 206 206 + tokio::time::advance(Duration::from_secs(5)).await; 207 207 + assert_eq!( 208 208 + a.now_unix_micros(), 209 209 + b.now_unix_micros(), 210 210 + "two sim clocks anchored at the same base must agree on virtual time", 142 211 ); 143 212 } 144 213

+66

crates/runtime/src/entropy.rs

Reviewed

··· 1 1 + use std::sync::atomic::{AtomicU64, Ordering}; 2 2 + 1 3 pub trait Entropy: Send + Sync + 'static { 2 4 fn next_u64(&self) -> u64; 3 5 } ··· 13 15 } 14 16 } 15 17 18 18 + #[derive(Debug)] 19 19 + pub struct SeededEntropy { 20 20 + state: AtomicU64, 21 21 + } 22 22 + 23 23 + impl SeededEntropy { 24 24 + const GOLDEN: u64 = 0x9E37_79B9_7F4A_7C15; 25 25 + 26 26 + pub fn new(seed: u64) -> Self { 27 27 + Self { 28 28 + state: AtomicU64::new(seed), 29 29 + } 30 30 + } 31 31 + } 32 32 + 33 33 + impl Entropy for SeededEntropy { 34 34 + fn next_u64(&self) -> u64 { 35 35 + let prev = self.state.fetch_add(Self::GOLDEN, Ordering::Relaxed); 36 36 + let z = prev.wrapping_add(Self::GOLDEN); 37 37 + let z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); 38 38 + let z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); 39 39 + z ^ (z >> 31) 40 40 + } 41 41 + } 42 42 + 16 43 #[cfg(test)] 17 44 mod tests { 18 45 use super::*; ··· 32 59 let e: Arc<dyn Entropy> = Arc::new(CountingEntropy(AtomicU64::new(100))); 33 60 assert_eq!(e.next_u64(), 100); 34 61 assert_eq!(e.next_u64(), 101); 62 62 + } 63 63 + 64 64 + #[test] 65 65 + fn seeded_entropy_two_constructions_with_same_seed_yield_same_stream() { 66 66 + let a = SeededEntropy::new(0xCAFE_F00D); 67 67 + let b = SeededEntropy::new(0xCAFE_F00D); 68 68 + for _ in 0..32 { 69 69 + assert_eq!( 70 70 + a.next_u64(), 71 71 + b.next_u64(), 72 72 + "splitmix stream must reproduce exactly across constructions with same seed", 73 73 + ); 74 74 + } 75 75 + } 76 76 + 77 77 + #[test] 78 78 + fn seeded_entropy_different_seeds_diverge() { 79 79 + let a = SeededEntropy::new(1); 80 80 + let b = SeededEntropy::new(2); 81 81 + let mut all_match = true; 82 82 + for _ in 0..32 { 83 83 + if a.next_u64() != b.next_u64() { 84 84 + all_match = false; 85 85 + break; 86 86 + } 87 87 + } 88 88 + assert!( 89 89 + !all_match, 90 90 + "two seeded entropies with distinct seeds must diverge inside 32 draws", 91 91 + ); 92 92 + } 93 93 + 94 94 + #[test] 95 95 + fn seeded_entropy_does_not_repeat_inside_short_window() { 96 96 + let e = SeededEntropy::new(0); 97 97 + let mut seen = std::collections::HashSet::new(); 98 98 + for _ in 0..1024 { 99 99 + assert!(seen.insert(e.next_u64()), "splitmix collided inside 1024 draws"); 100 100 + } 35 101 } 36 102 37 103 #[test]

+7 -2

crates/runtime/src/lib.rs

Reviewed

··· 1 1 mod clock; 2 2 mod entropy; 3 3 mod hasher; 4 4 + mod mem_network; 4 5 mod network; 5 6 6 6 - pub use clock::{Clock, SleepFuture, SystemClock, UnixMicros}; 7 7 - pub use entropy::{Entropy, OsEntropy}; 7 7 + pub use clock::{Clock, SimClock, SleepFuture, SystemClock, UnixMicros}; 8 8 + pub use entropy::{Entropy, OsEntropy, SeededEntropy}; 8 9 pub use hasher::RuntimeHasher; 10 10 + pub use mem_network::{ 11 11 + DEFAULT_MEM_WS_CAPACITY, MemHttpBody, MemHttpResponder, MemHttpResponse, MemHttpTransport, 12 12 + MemWsResponder, MemWsServerFuture, MemWsTransport, 13 13 + }; 9 14 pub use network::{ 10 15 BodyStream, HttpRequest, HttpResponseFuture, HttpResponseHead, HttpResult, HttpTransport, 11 16 NetworkError, ReqwestHttp, TungsteniteWs, WsConn, WsConnectFuture, WsMessage, WsMessageFuture,

+352

crates/runtime/src/mem_network.rs

Reviewed

··· 1 1 + use std::future::Future; 2 2 + use std::pin::Pin; 3 3 + use std::sync::Arc; 4 4 + use std::time::Duration; 5 5 + 6 6 + use bytes::Bytes; 7 7 + use futures::stream; 8 8 + use http::{HeaderMap, StatusCode}; 9 9 + use tokio::sync::mpsc; 10 10 + use url::Url; 11 11 + 12 12 + use crate::clock::{Clock, SleepFuture}; 13 13 + use crate::network::{ 14 14 + BodyStream, HttpRequest, HttpResponseFuture, HttpResponseHead, HttpResult, HttpTransport, 15 15 + NetworkError, WsConn, WsConnectFuture, WsMessage, WsMessageFuture, WsSendFuture, WsSink, 16 16 + WsStream, WsTransport, 17 17 + }; 18 18 + 19 19 + #[derive(Debug)] 20 20 + pub struct MemHttpResponse { 21 21 + pub latency: Duration, 22 22 + pub result: Result<MemHttpBody, NetworkError>, 23 23 + } 24 24 + 25 25 + #[derive(Debug)] 26 26 + pub struct MemHttpBody { 27 27 + pub status: StatusCode, 28 28 + pub headers: HeaderMap, 29 29 + pub body: Bytes, 30 30 + } 31 31 + 32 32 + impl MemHttpBody { 33 33 + pub fn ok_json(body: Bytes) -> Self { 34 34 + let mut headers = HeaderMap::new(); 35 35 + headers.insert(http::header::CONTENT_TYPE, "application/json".parse().unwrap()); 36 36 + Self { 37 37 + status: StatusCode::OK, 38 38 + headers, 39 39 + body, 40 40 + } 41 41 + } 42 42 + 43 43 + pub fn status_only(status: StatusCode) -> Self { 44 44 + Self { 45 45 + status, 46 46 + headers: HeaderMap::new(), 47 47 + body: Bytes::new(), 48 48 + } 49 49 + } 50 50 + } 51 51 + 52 52 + pub trait MemHttpResponder: Send + Sync + 'static { 53 53 + fn respond(&self, request: &HttpRequest) -> MemHttpResponse; 54 54 + } 55 55 + 56 56 + #[derive(Clone)] 57 57 + pub struct MemHttpTransport { 58 58 + responder: Arc<dyn MemHttpResponder>, 59 59 + clock: Arc<dyn Clock>, 60 60 + } 61 61 + 62 62 + impl MemHttpTransport { 63 63 + pub fn new(responder: Arc<dyn MemHttpResponder>, clock: Arc<dyn Clock>) -> Self { 64 64 + Self { responder, clock } 65 65 + } 66 66 + 67 67 + pub fn shared(responder: Arc<dyn MemHttpResponder>, clock: Arc<dyn Clock>) -> Arc<dyn HttpTransport> { 68 68 + Arc::new(Self::new(responder, clock)) 69 69 + } 70 70 + } 71 71 + 72 72 + impl HttpTransport for MemHttpTransport { 73 73 + fn execute(&self, request: HttpRequest) -> HttpResponseFuture { 74 74 + let response = self.responder.respond(&request); 75 75 + let sleep: SleepFuture = self.clock.sleep(response.latency); 76 76 + Box::pin(async move { 77 77 + sleep.await; 78 78 + into_response_head(response.result) 79 79 + }) 80 80 + } 81 81 + } 82 82 + 83 83 + fn into_response_head(result: Result<MemHttpBody, NetworkError>) -> HttpResult { 84 84 + let body = result?; 85 85 + let content_length = Some(body.body.len() as u64); 86 86 + let body_stream: BodyStream = Box::pin(stream::once(async move { Ok(body.body) })); 87 87 + Ok(HttpResponseHead { 88 88 + status: body.status, 89 89 + headers: body.headers, 90 90 + content_length, 91 91 + body: body_stream, 92 92 + }) 93 93 + } 94 94 + 95 95 + pub type MemWsServerFuture = Pin<Box<dyn Future<Output = ()> + Send + 'static>>; 96 96 + 97 97 + pub trait MemWsResponder: Send + Sync + 'static { 98 98 + fn spawn_server( 99 99 + &self, 100 100 + url: Url, 101 101 + recv: mpsc::UnboundedReceiver<WsMessage>, 102 102 + send: mpsc::Sender<WsMessage>, 103 103 + ) -> MemWsServerFuture; 104 104 + } 105 105 + 106 106 + pub const DEFAULT_MEM_WS_CAPACITY: usize = 4096; 107 107 + 108 108 + #[derive(Clone)] 109 109 + pub struct MemWsTransport { 110 110 + responder: Arc<dyn MemWsResponder>, 111 111 + capacity: usize, 112 112 + } 113 113 + 114 114 + impl MemWsTransport { 115 115 + pub fn new(responder: Arc<dyn MemWsResponder>) -> Self { 116 116 + Self::with_capacity(responder, DEFAULT_MEM_WS_CAPACITY) 117 117 + } 118 118 + 119 119 + pub fn with_capacity(responder: Arc<dyn MemWsResponder>, capacity: usize) -> Self { 120 120 + assert!(capacity > 0, "MemWsTransport capacity must be > 0"); 121 121 + Self { 122 122 + responder, 123 123 + capacity, 124 124 + } 125 125 + } 126 126 + 127 127 + pub fn shared(responder: Arc<dyn MemWsResponder>) -> Arc<dyn WsTransport> { 128 128 + Arc::new(Self::new(responder)) 129 129 + } 130 130 + 131 131 + pub fn shared_with_capacity( 132 132 + responder: Arc<dyn MemWsResponder>, 133 133 + capacity: usize, 134 134 + ) -> Arc<dyn WsTransport> { 135 135 + Arc::new(Self::with_capacity(responder, capacity)) 136 136 + } 137 137 + } 138 138 + 139 139 + impl WsTransport for MemWsTransport { 140 140 + fn connect(&self, url: Url) -> WsConnectFuture { 141 141 + let responder = self.responder.clone(); 142 142 + let capacity = self.capacity; 143 143 + Box::pin(async move { 144 144 + let (c2s_tx, c2s_rx) = mpsc::unbounded_channel(); 145 145 + let (s2c_tx, s2c_rx) = mpsc::channel(capacity); 146 146 + let server_future = responder.spawn_server(url, c2s_rx, s2c_tx); 147 147 + tokio::spawn(server_future); 148 148 + let sink: Box<dyn WsSink> = Box::new(MemWsSink { sender: c2s_tx }); 149 149 + let stream: Box<dyn WsStream> = Box::new(MemWsStream { receiver: s2c_rx }); 150 150 + Ok(WsConn { sink, stream }) 151 151 + }) 152 152 + } 153 153 + } 154 154 + 155 155 + struct MemWsSink { 156 156 + sender: mpsc::UnboundedSender<WsMessage>, 157 157 + } 158 158 + 159 159 + impl WsSink for MemWsSink { 160 160 + fn send<'a>(&'a mut self, message: WsMessage) -> WsSendFuture<'a> { 161 161 + let result = self 162 162 + .sender 163 163 + .send(message) 164 164 + .map_err(|_| NetworkError::Transport("server side closed channel".into())); 165 165 + Box::pin(async move { result }) 166 166 + } 167 167 + } 168 168 + 169 169 + struct MemWsStream { 170 170 + receiver: mpsc::Receiver<WsMessage>, 171 171 + } 172 172 + 173 173 + impl WsStream for MemWsStream { 174 174 + fn next<'a>(&'a mut self) -> WsMessageFuture<'a> { 175 175 + Box::pin(async move { self.receiver.recv().await.map(Ok) }) 176 176 + } 177 177 + } 178 178 + 179 179 + #[cfg(test)] 180 180 + mod tests { 181 181 + use super::*; 182 182 + use crate::SimClock; 183 183 + use crate::UnixMicros; 184 184 + use std::sync::Mutex; 185 185 + use std::sync::atomic::{AtomicUsize, Ordering}; 186 186 + use tokio::time::Instant; 187 187 + 188 188 + struct ScriptedHttp { 189 189 + responses: Mutex<Vec<MemHttpResponse>>, 190 190 + cursor: AtomicUsize, 191 191 + } 192 192 + 193 193 + impl ScriptedHttp { 194 194 + fn new(responses: Vec<MemHttpResponse>) -> Self { 195 195 + Self { 196 196 + responses: Mutex::new(responses), 197 197 + cursor: AtomicUsize::new(0), 198 198 + } 199 199 + } 200 200 + } 201 201 + 202 202 + impl MemHttpResponder for ScriptedHttp { 203 203 + fn respond(&self, _: &HttpRequest) -> MemHttpResponse { 204 204 + let i = self.cursor.fetch_add(1, Ordering::Relaxed); 205 205 + let mut guard = self.responses.lock().unwrap(); 206 206 + let next = std::mem::replace( 207 207 + &mut guard[i], 208 208 + MemHttpResponse { 209 209 + latency: Duration::ZERO, 210 210 + result: Err(NetworkError::Transport("script consumed".into())), 211 211 + }, 212 212 + ); 213 213 + next 214 214 + } 215 215 + } 216 216 + 217 217 + fn ok_response(body: &str, latency_ms: u64) -> MemHttpResponse { 218 218 + MemHttpResponse { 219 219 + latency: Duration::from_millis(latency_ms), 220 220 + result: Ok(MemHttpBody::ok_json(Bytes::from(body.to_owned()))), 221 221 + } 222 222 + } 223 223 + 224 224 + fn err_response(latency_ms: u64) -> MemHttpResponse { 225 225 + MemHttpResponse { 226 226 + latency: Duration::from_millis(latency_ms), 227 227 + result: Err(NetworkError::Transport("brownout".into())), 228 228 + } 229 229 + } 230 230 + 231 231 + #[tokio::test(start_paused = true)] 232 232 + async fn http_returns_scripted_body_after_injected_latency() { 233 233 + let clock: Arc<dyn Clock> = Arc::new(SimClock::at(UnixMicros::new(0))); 234 234 + let responder: Arc<dyn MemHttpResponder> = Arc::new(ScriptedHttp::new(vec![ 235 235 + ok_response("{\"hello\":1}", 10), 236 236 + ])); 237 237 + let transport = MemHttpTransport::new(responder, clock); 238 238 + let request = HttpRequest { 239 239 + url: Url::parse("http://oyster.cafe/xrpc/x").unwrap(), 240 240 + headers: HeaderMap::new(), 241 241 + }; 242 242 + 243 243 + let before = Instant::now(); 244 244 + let resp = transport.execute(request).await.unwrap(); 245 245 + let elapsed = Instant::now().saturating_duration_since(before); 246 246 + 247 247 + assert_eq!(resp.status, StatusCode::OK); 248 248 + assert_eq!(elapsed, Duration::from_millis(10)); 249 249 + let chunks = collect_body(resp.body).await; 250 250 + assert_eq!(chunks.as_ref(), b"{\"hello\":1}"); 251 251 + } 252 252 + 253 253 + #[tokio::test(start_paused = true)] 254 254 + async fn http_propagates_scripted_errors_with_latency() { 255 255 + let clock: Arc<dyn Clock> = Arc::new(SimClock::at(UnixMicros::new(0))); 256 256 + let responder: Arc<dyn MemHttpResponder> = 257 257 + Arc::new(ScriptedHttp::new(vec![err_response(50)])); 258 258 + let transport = MemHttpTransport::new(responder, clock); 259 259 + let request = HttpRequest { 260 260 + url: Url::parse("http://oyster.cafe/xrpc/x").unwrap(), 261 261 + headers: HeaderMap::new(), 262 262 + }; 263 263 + 264 264 + let before = Instant::now(); 265 265 + let resp = transport.execute(request).await; 266 266 + let elapsed = Instant::now().saturating_duration_since(before); 267 267 + 268 268 + assert_eq!(elapsed, Duration::from_millis(50)); 269 269 + assert!(matches!(resp, Err(NetworkError::Transport(_)))); 270 270 + } 271 271 + 272 272 + async fn collect_body(mut body: BodyStream) -> Bytes { 273 273 + use futures::StreamExt; 274 274 + let mut acc = Vec::new(); 275 275 + while let Some(chunk) = body.next().await { 276 276 + acc.extend_from_slice(&chunk.unwrap()); 277 277 + } 278 278 + Bytes::from(acc) 279 279 + } 280 280 + 281 281 + struct ScriptedWs { 282 282 + frames: Mutex<Vec<WsMessage>>, 283 283 + } 284 284 + 285 285 + impl MemWsResponder for ScriptedWs { 286 286 + fn spawn_server( 287 287 + &self, 288 288 + _: Url, 289 289 + mut _recv: mpsc::UnboundedReceiver<WsMessage>, 290 290 + send: mpsc::Sender<WsMessage>, 291 291 + ) -> MemWsServerFuture { 292 292 + let frames: Vec<WsMessage> = std::mem::take(&mut *self.frames.lock().unwrap()); 293 293 + Box::pin(async move { 294 294 + for frame in frames { 295 295 + if send.send(frame).await.is_err() { 296 296 + return; 297 297 + } 298 298 + } 299 299 + }) 300 300 + } 301 301 + } 302 302 + 303 303 + #[tokio::test(start_paused = true)] 304 304 + async fn ws_delivers_scripted_frames_to_client() { 305 305 + let responder: Arc<dyn MemWsResponder> = Arc::new(ScriptedWs { 306 306 + frames: Mutex::new(vec![ 307 307 + WsMessage::Text("frame-a".into()), 308 308 + WsMessage::Text("frame-b".into()), 309 309 + ]), 310 310 + }); 311 311 + let transport = MemWsTransport::new(responder); 312 312 + let mut conn = transport 313 313 + .connect(Url::parse("ws://oyster.cafe/").unwrap()) 314 314 + .await 315 315 + .unwrap(); 316 316 + let a = conn.stream.next().await.unwrap().unwrap(); 317 317 + let b = conn.stream.next().await.unwrap().unwrap(); 318 318 + assert!(matches!(a, WsMessage::Text(t) if t == "frame-a")); 319 319 + assert!(matches!(b, WsMessage::Text(t) if t == "frame-b")); 320 320 + } 321 321 + 322 322 + struct EchoWs; 323 323 + 324 324 + impl MemWsResponder for EchoWs { 325 325 + fn spawn_server( 326 326 + &self, 327 327 + _: Url, 328 328 + mut recv: mpsc::UnboundedReceiver<WsMessage>, 329 329 + send: mpsc::Sender<WsMessage>, 330 330 + ) -> MemWsServerFuture { 331 331 + Box::pin(async move { 332 332 + while let Some(msg) = recv.recv().await { 333 333 + if send.send(msg).await.is_err() { 334 334 + return; 335 335 + } 336 336 + } 337 337 + }) 338 338 + } 339 339 + } 340 340 + 341 341 + #[tokio::test(start_paused = true)] 342 342 + async fn ws_round_trip_via_server_echo() { 343 343 + let transport = MemWsTransport::new(Arc::new(EchoWs)); 344 344 + let mut conn = transport 345 345 + .connect(Url::parse("ws://oyster.cafe/").unwrap()) 346 346 + .await 347 347 + .unwrap(); 348 348 + conn.sink.send(WsMessage::Text("ping".into())).await.unwrap(); 349 349 + let echoed = conn.stream.next().await.unwrap().unwrap(); 350 350 + assert!(matches!(echoed, WsMessage::Text(t) if t == "ping")); 351 351 + } 352 352 + }