fetch_bounds.rs (9108B)
1 use futures::{SinkExt, StreamExt}; 2 use nostr_sdk::prelude::{EventBuilder, JsonUtil, Keys}; 3 use radroots_transport::{ 4 EventSource, FetchRequest, TargetSet, outcome::FetchTargetState, source::FetchBounds, 5 }; 6 use radroots_transport_nostr::{ 7 Config, NostrTransport, RelayAccess, RelayEndpoint, RelayProfile, RelayProfileKind, 8 RelayUrlPolicy, 9 }; 10 use serde_json::Value; 11 use std::{ 12 sync::{ 13 Arc, 14 atomic::{AtomicUsize, Ordering}, 15 }, 16 time::{Duration, SystemTime, UNIX_EPOCH}, 17 }; 18 use tokio::{net::TcpListener, sync::oneshot}; 19 use tokio_tungstenite::{accept_async, tungstenite::Message}; 20 21 fn transport(urls: &[String], connections: usize) -> (NostrTransport, FetchRequest) { 22 let endpoints = urls 23 .iter() 24 .map(|url| RelayEndpoint::new(url, RelayUrlPolicy::Local, RelayAccess::ReadOnly).unwrap()); 25 let profile = RelayProfile::explicit(RelayProfileKind::Simulator, endpoints).unwrap(); 26 let config = Config::from_profile(profile) 27 .with_timeouts(1000, 1000, 500) 28 .unwrap() 29 .with_max_connections(connections) 30 .unwrap(); 31 let targets = TargetSet::new( 32 config 33 .read_relays() 34 .map(|relay| relay.to_target().unwrap()) 35 .collect(), 36 ) 37 .unwrap(); 38 let now = SystemTime::now() 39 .duration_since(UNIX_EPOCH) 40 .unwrap() 41 .as_millis() as u64; 42 let request = FetchRequest::new( 43 "bounded-loopback", 44 targets, 45 FetchBounds::new(10, now + 5000).unwrap(), 46 ) 47 .unwrap(); 48 (NostrTransport::new(config), request) 49 } 50 51 async fn listener() -> (TcpListener, String) { 52 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); 53 let url = format!("ws://{}", listener.local_addr().unwrap()); 54 (listener, url) 55 } 56 57 fn event(content: &str) -> Value { 58 let keys = 59 Keys::parse("0000000000000000000000000000000000000000000000000000000000000001").unwrap(); 60 let event = EventBuilder::text_note(content) 61 .sign_with_keys(&keys) 62 .unwrap(); 63 serde_json::from_str(&event.as_json()).unwrap() 64 } 65 66 async fn serve( 67 listener: TcpListener, 68 payload: Option<Value>, 69 eose: bool, 70 requests: Arc<AtomicUsize>, 71 ) { 72 let (stream, _) = listener.accept().await.unwrap(); 73 let mut socket = accept_async(stream).await.unwrap(); 74 while let Some(message) = socket.next().await { 75 let Ok(Message::Text(message)) = message else { 76 continue; 77 }; 78 let values: Value = serde_json::from_str(&message).unwrap(); 79 if values[0] != "REQ" { 80 continue; 81 } 82 requests.fetch_add(1, Ordering::SeqCst); 83 if let Some(payload) = &payload { 84 socket 85 .send(Message::Text( 86 serde_json::to_string(&("EVENT", &values[1], payload)) 87 .unwrap() 88 .into(), 89 )) 90 .await 91 .unwrap(); 92 } 93 if eose { 94 socket 95 .send(Message::Text( 96 serde_json::to_string(&("EOSE", &values[1])).unwrap().into(), 97 )) 98 .await 99 .unwrap(); 100 } 101 std::future::pending::<()>().await; 102 } 103 } 104 105 #[tokio::test(flavor = "multi_thread")] 106 async fn queued_relays_do_not_receive_a_fresh_timeout_after_a_stalled_relay() { 107 let (first, first_url) = listener().await; 108 let (second, second_url) = listener().await; 109 // Adapter scheduling is canonical URL order, independent of profile order. 110 let mut relays = [(first_url, first), (second_url, second)]; 111 relays.sort_by(|left, right| left.0.cmp(&right.0)); 112 let [(first_url, first), (second_url, second)] = relays; 113 let first_requests = Arc::new(AtomicUsize::new(0)); 114 let second_requests = Arc::new(AtomicUsize::new(0)); 115 let first_task = tokio::spawn(serve(first, None, false, Arc::clone(&first_requests))); 116 let second_task = tokio::spawn(serve(second, None, true, Arc::clone(&second_requests))); 117 let (transport, request) = transport(&[first_url, second_url], 1); 118 let page = tokio::time::timeout(Duration::from_secs(5), transport.fetch(request)) 119 .await 120 .unwrap() 121 .unwrap(); 122 first_task.abort(); 123 second_task.abort(); 124 assert_eq!(first_requests.load(Ordering::SeqCst), 1); 125 assert_eq!(second_requests.load(Ordering::SeqCst), 0); 126 assert_eq!(page.target_outcomes().len(), 2); 127 assert!( 128 page.target_outcomes() 129 .iter() 130 .all(|outcome| outcome.state() == FetchTargetState::Cancelled) 131 ); 132 } 133 134 #[tokio::test(flavor = "multi_thread")] 135 async fn a_slow_relay_preserves_the_other_relays_completed_evidence_and_collected_events() { 136 let (good, good_url) = listener().await; 137 let (slow, slow_url) = listener().await; 138 let good_event = event("complete relay"); 139 let slow_event = event("partial relay"); 140 let expected = [ 141 good_event["id"].as_str().unwrap().to_owned(), 142 slow_event["id"].as_str().unwrap().to_owned(), 143 ]; 144 let calls = Arc::new(AtomicUsize::new(0)); 145 let good_task = tokio::spawn(serve(good, Some(good_event), true, Arc::clone(&calls))); 146 let slow_task = tokio::spawn(serve(slow, Some(slow_event), false, Arc::clone(&calls))); 147 let (transport, request) = transport(&[good_url, slow_url], 2); 148 let page = tokio::time::timeout(Duration::from_secs(5), transport.fetch(request)) 149 .await 150 .unwrap() 151 .unwrap(); 152 good_task.abort(); 153 slow_task.abort(); 154 assert_eq!(calls.load(Ordering::SeqCst), 2); 155 assert_eq!(page.events().len(), 2); 156 for id in expected { 157 assert!( 158 page.events() 159 .iter() 160 .any(|event| event.event().id_str() == id) 161 ); 162 } 163 assert_eq!( 164 page.target_outcomes() 165 .iter() 166 .filter(|outcome| outcome.state() == FetchTargetState::Complete) 167 .count(), 168 1 169 ); 170 assert_eq!( 171 page.target_outcomes() 172 .iter() 173 .filter(|outcome| outcome.state() == FetchTargetState::Cancelled) 174 .count(), 175 1 176 ); 177 } 178 179 #[tokio::test(flavor = "multi_thread")] 180 async fn oversized_wire_messages_never_become_completed_fetch_evidence() { 181 let (listener, url) = listener().await; 182 let server = tokio::spawn(async move { 183 let (stream, _) = listener.accept().await.unwrap(); 184 let mut socket = accept_async(stream).await.unwrap(); 185 while let Some(message) = socket.next().await { 186 let message = match message { 187 Ok(Message::Text(message)) => message, 188 Ok(Message::Close(_)) | Err(_) => return, 189 _ => continue, 190 }; 191 let values: Value = serde_json::from_str(&message).unwrap(); 192 if values[0] == "REQ" { 193 let _ = socket 194 .send(Message::Text( 195 serde_json::to_string(&("NOTICE", "x".repeat(512 * 1024))) 196 .unwrap() 197 .into(), 198 )) 199 .await; 200 } 201 } 202 }); 203 let (transport, request) = transport(&[url], 1); 204 let page = tokio::time::timeout(Duration::from_secs(5), transport.fetch(request)) 205 .await 206 .unwrap() 207 .unwrap(); 208 tokio::time::timeout(Duration::from_secs(5), server) 209 .await 210 .unwrap() 211 .unwrap(); 212 assert!(page.events().is_empty()); 213 assert_eq!(page.target_outcomes().len(), 1); 214 assert_ne!( 215 page.target_outcomes()[0].state(), 216 FetchTargetState::Complete 217 ); 218 } 219 220 #[tokio::test(flavor = "multi_thread")] 221 async fn dropping_a_polled_fetch_closes_the_subscription_or_connection() { 222 let (listener, url) = listener().await; 223 let (started, observed) = oneshot::channel(); 224 let server = tokio::spawn(async move { 225 let (stream, _) = listener.accept().await.unwrap(); 226 let mut socket = accept_async(stream).await.unwrap(); 227 let mut started = Some(started); 228 while let Some(message) = socket.next().await { 229 let message = match message { 230 Ok(Message::Text(message)) => message, 231 Ok(Message::Close(_)) | Err(_) => return, 232 _ => continue, 233 }; 234 let values: Value = serde_json::from_str(&message).unwrap(); 235 if values[0] == "REQ" { 236 started.take().unwrap().send(()).unwrap(); 237 } else if values[0] == "CLOSE" { 238 return; 239 } 240 } 241 }); 242 let (transport, request) = transport(&[url], 1); 243 let mut fetch = Box::pin(transport.fetch(request)); 244 tokio::time::timeout(Duration::from_secs(5), async { 245 tokio::select! { 246 _ = &mut fetch => panic!("fetch completed before its relay response"), 247 result = observed => { result.unwrap(); } 248 } 249 }) 250 .await 251 .unwrap(); 252 drop(fetch); 253 tokio::time::timeout(Duration::from_secs(5), server) 254 .await 255 .unwrap() 256 .unwrap(); 257 }