subscription_contract.rs (14436B)
1 use core::{future::Future, pin::Pin, task::Context}; 2 use std::sync::{ 3 Arc, 4 atomic::{AtomicBool, Ordering}, 5 }; 6 7 use futures::{executor::block_on, future, task::noop_waker_ref}; 8 use radroots_event::{SignedEvent, wire::v1::Nip01EventWire}; 9 use radroots_transport::{ 10 BoxFuture, BoxSubscription, Error, EventSubscriber, EventSubscription, SubscriptionEnd, 11 SubscriptionEndReason, SubscriptionEvent, SubscriptionNext, SubscriptionRequest, Target, 12 TargetSet, TransportId, 13 source::{ 14 EventProvenance, FetchCursor, FetchSelector, ObservedEvent, SUBSCRIPTION_MAX_EVENTS, 15 SUBSCRIPTION_REQUEST_ID_MAX_BYTES, SubscriptionBounds, SubscriptionCheckpoint, 16 SubscriptionRequestId, 17 }, 18 }; 19 20 fn target(uri: &str) -> Target { 21 Target::nostr_relay(uri).expect("nostr target") 22 } 23 24 fn signed_event() -> SignedEvent { 25 let raw = r#"{"id":"56bfc78223bb2221bad82b539efdec1ade0f56d0eb0e1f592fd387df4b2ceee0","pubkey":"585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df","created_at":1700000001,"kind":0,"tags":[],"content":"{}","sig":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"}"#; 26 let wire = Nip01EventWire::parse_json(raw).expect("wire event"); 27 SignedEvent::from_wire_verified_id(wire, raw).expect("signed event") 28 } 29 30 fn request(targets: TargetSet, limit: u16) -> SubscriptionRequest { 31 SubscriptionRequest::new( 32 "subscription-request", 33 targets, 34 SubscriptionBounds::new(limit, 1_700_000_100_000).expect("bounds"), 35 ) 36 .expect("request") 37 } 38 39 #[test] 40 fn subscription_identity_and_bounds_are_exact_and_bounded() { 41 assert_eq!( 42 SubscriptionRequestId::parse("").expect_err("empty id"), 43 Error::EmptySubscriptionRequestId 44 ); 45 for invalid in [" request", "request ", "request\nid"] { 46 assert_eq!( 47 SubscriptionRequestId::parse(invalid).expect_err("invalid id"), 48 Error::InvalidSubscriptionRequestId 49 ); 50 } 51 assert_eq!( 52 SubscriptionRequestId::parse("x".repeat(SUBSCRIPTION_REQUEST_ID_MAX_BYTES + 1)) 53 .expect_err("oversized id"), 54 Error::InvalidSubscriptionRequestId 55 ); 56 let maximum = SubscriptionRequestId::parse("x".repeat(SUBSCRIPTION_REQUEST_ID_MAX_BYTES)) 57 .expect("maximum id"); 58 assert_eq!(maximum.as_str().len(), SUBSCRIPTION_REQUEST_ID_MAX_BYTES); 59 assert_eq!(maximum.to_string(), maximum.as_str()); 60 61 assert_eq!( 62 SubscriptionBounds::new(0, 1).expect_err("zero limit"), 63 Error::InvalidSubscriptionLimit 64 ); 65 assert_eq!( 66 SubscriptionBounds::new(SUBSCRIPTION_MAX_EVENTS + 1, 1).expect_err("oversized limit"), 67 Error::InvalidSubscriptionLimit 68 ); 69 assert_eq!( 70 SubscriptionBounds::new(1, 0).expect_err("zero deadline"), 71 Error::InvalidSubscriptionDeadline 72 ); 73 let maximum = 74 SubscriptionBounds::new(SUBSCRIPTION_MAX_EVENTS, u64::MAX).expect("maximum bounds"); 75 assert_eq!(maximum.event_limit(), SUBSCRIPTION_MAX_EVENTS); 76 assert_eq!(maximum.deadline_unix_ms(), u64::MAX); 77 } 78 79 #[test] 80 fn checkpoints_are_bounded_unique_and_canonical_for_the_target_set() { 81 let first = target("wss://one.example"); 82 let second = target("wss://two.example"); 83 let targets = TargetSet::new(vec![first.clone(), second.clone()]).expect("targets"); 84 let first_checkpoint = SubscriptionCheckpoint::new( 85 first.fingerprint().clone(), 86 FetchCursor::parse("first").expect("cursor"), 87 ); 88 let second_checkpoint = SubscriptionCheckpoint::new( 89 second.fingerprint().clone(), 90 FetchCursor::parse("second").expect("cursor"), 91 ); 92 let configured = request(targets.clone(), 2) 93 .with_checkpoints([second_checkpoint.clone(), first_checkpoint.clone()]) 94 .expect("checkpoints"); 95 assert_eq!( 96 configured.checkpoints(), 97 &[first_checkpoint.clone(), second_checkpoint] 98 ); 99 100 assert_eq!( 101 request(targets.clone(), 2) 102 .with_checkpoints([first_checkpoint.clone(), first_checkpoint.clone()]) 103 .expect_err("duplicate"), 104 Error::DuplicateSubscriptionCheckpoint 105 ); 106 let foreign = target("wss://foreign.example"); 107 assert_eq!( 108 request(targets.clone(), 2) 109 .with_checkpoints([SubscriptionCheckpoint::new( 110 foreign.fingerprint().clone(), 111 FetchCursor::parse("foreign").expect("cursor"), 112 )]) 113 .expect_err("foreign"), 114 Error::UnexpectedSubscriptionCheckpoint 115 ); 116 assert_eq!( 117 request(TargetSet::new(vec![first]).expect("targets"), 1) 118 .with_checkpoints(core::iter::repeat(first_checkpoint)) 119 .expect_err("infinite iterator is bounded"), 120 Error::SubscriptionCheckpointSetTooLarge 121 ); 122 } 123 124 #[test] 125 fn live_events_bind_selector_target_transport_and_checkpoint() { 126 let requested = target("wss://one.example"); 127 let targets = TargetSet::new(vec![requested.clone()]).expect("targets"); 128 let request = request(targets, 2) 129 .with_selector(FetchSelector::all().with_kinds(vec![0]).expect("selector")); 130 let cursor = FetchCursor::parse("event-1").expect("cursor"); 131 let observed = ObservedEvent::new( 132 signed_event(), 133 EventProvenance::new( 134 TransportId::NOSTR, 135 requested.fingerprint().clone(), 136 1_700_000_000_001, 137 ) 138 .expect("provenance") 139 .with_cursor(cursor.clone()), 140 ); 141 let event = SubscriptionEvent::for_request( 142 &request, 143 observed.clone(), 144 SubscriptionCheckpoint::new(requested.fingerprint().clone(), cursor), 145 ) 146 .expect("event"); 147 event 148 .validate_for_request(&request) 149 .expect("request binding"); 150 assert_eq!(event.request_id(), request.request_id()); 151 assert_eq!(event.observed().event().id_str(), signed_event().id_str()); 152 153 assert_eq!( 154 SubscriptionEvent::for_request( 155 &request, 156 observed, 157 SubscriptionCheckpoint::new( 158 requested.fingerprint().clone(), 159 FetchCursor::parse("different").expect("cursor"), 160 ), 161 ) 162 .expect_err("cursor mismatch"), 163 Error::SubscriptionEventCheckpointMismatch 164 ); 165 166 let filtered = 167 request.with_selector(FetchSelector::all().with_kinds(vec![1]).expect("selector")); 168 let observed = ObservedEvent::new( 169 signed_event(), 170 EventProvenance::new(TransportId::NOSTR, requested.fingerprint().clone(), 1) 171 .expect("provenance") 172 .with_cursor(FetchCursor::parse("event-2").expect("cursor")), 173 ); 174 assert_eq!( 175 SubscriptionEvent::for_request( 176 &filtered, 177 observed, 178 SubscriptionCheckpoint::new( 179 requested.fingerprint().clone(), 180 FetchCursor::parse("event-2").expect("cursor"), 181 ), 182 ) 183 .expect_err("selector mismatch"), 184 Error::UnexpectedSubscriptionEvent 185 ); 186 } 187 188 struct StableSubscription { 189 request: SubscriptionRequest, 190 terminal: SubscriptionEnd, 191 } 192 193 impl EventSubscription for StableSubscription { 194 fn request(&self) -> &SubscriptionRequest { 195 &self.request 196 } 197 198 fn next(&mut self) -> BoxFuture<'_, Result<SubscriptionNext, Error>> { 199 let terminal = self.terminal.clone(); 200 Box::pin(async move { Ok(SubscriptionNext::End(terminal)) }) 201 } 202 203 fn cancel(&mut self) -> BoxFuture<'_, Result<SubscriptionEnd, Error>> { 204 let terminal = self.terminal.clone(); 205 Box::pin(async move { Ok(terminal) }) 206 } 207 } 208 209 struct StableSubscriber; 210 211 impl EventSubscriber for StableSubscriber { 212 fn subscribe( 213 &self, 214 request: SubscriptionRequest, 215 ) -> BoxFuture<'_, Result<BoxSubscription, Error>> { 216 Box::pin(async move { 217 let terminal = 218 SubscriptionEnd::for_request(&request, 0, [], SubscriptionEndReason::SourceClosed)?; 219 Ok(Box::new(StableSubscription { request, terminal }) as BoxSubscription) 220 }) 221 } 222 } 223 224 #[test] 225 fn subscription_spi_is_dyn_compatible_and_terminal_results_are_idempotent() { 226 let subscriber: &dyn EventSubscriber = &StableSubscriber; 227 let targets = TargetSet::new(vec![target("wss://one.example")]).expect("targets"); 228 let request = request(targets, 1); 229 let mut subscription = block_on(subscriber.subscribe(request.clone())).expect("subscribe"); 230 assert_eq!(subscription.request(), &request); 231 232 let first = block_on(subscription.next()).expect("next"); 233 let second = block_on(subscription.next()).expect("next again"); 234 let cancelled = block_on(subscription.cancel()).expect("cancel after end"); 235 assert_eq!(first, second); 236 assert_eq!(first, SubscriptionNext::End(cancelled.clone())); 237 assert_eq!(cancelled.reason(), SubscriptionEndReason::SourceClosed); 238 assert_eq!(cancelled.event_count(), 0); 239 assert!(cancelled.checkpoints().is_empty()); 240 cancelled 241 .validate_for_request(&request) 242 .expect("request-bound end"); 243 assert_eq!(cancelled.request(), &request); 244 245 assert_eq!( 246 SubscriptionEnd::for_request(&request, 2, [], SubscriptionEndReason::EventLimit,) 247 .expect_err("event limit exceeded"), 248 Error::SubscriptionEndLimitExceeded 249 ); 250 assert_eq!( 251 SubscriptionEnd::for_request(&request, 0, [], SubscriptionEndReason::EventLimit,) 252 .expect_err("event limit reason requires the exact limit"), 253 Error::InvalidSubscriptionEnd 254 ); 255 256 let other = SubscriptionRequest::new( 257 "other-request", 258 request.target_set().clone(), 259 request.bounds(), 260 ) 261 .expect("other request"); 262 assert_eq!( 263 cancelled 264 .validate_for_request(&other) 265 .expect_err("request mismatch"), 266 Error::SubscriptionEndRequestMismatch 267 ); 268 269 for (reason, event_count) in [ 270 (SubscriptionEndReason::EventLimit, 1), 271 (SubscriptionEndReason::Deadline, 0), 272 (SubscriptionEndReason::Cancelled, 0), 273 (SubscriptionEndReason::SourceClosed, 0), 274 ] { 275 assert_eq!( 276 SubscriptionEnd::for_request(&request, event_count, [], reason) 277 .expect("terminal reason") 278 .reason(), 279 reason 280 ); 281 } 282 } 283 284 struct CancellationGuard(Arc<AtomicBool>); 285 286 impl Drop for CancellationGuard { 287 fn drop(&mut self) { 288 self.0.store(true, Ordering::SeqCst); 289 } 290 } 291 292 struct PendingSubscription { 293 request: SubscriptionRequest, 294 terminal: SubscriptionEnd, 295 cancellation_observed: Arc<AtomicBool>, 296 } 297 298 impl EventSubscription for PendingSubscription { 299 fn request(&self) -> &SubscriptionRequest { 300 &self.request 301 } 302 303 fn next(&mut self) -> BoxFuture<'_, Result<SubscriptionNext, Error>> { 304 let cancellation_observed = Arc::clone(&self.cancellation_observed); 305 Box::pin(async move { 306 let _guard = CancellationGuard(cancellation_observed); 307 future::pending::<Result<SubscriptionNext, Error>>().await 308 }) 309 } 310 311 fn cancel(&mut self) -> BoxFuture<'_, Result<SubscriptionEnd, Error>> { 312 let terminal = self.terminal.clone(); 313 Box::pin(async move { Ok(terminal) }) 314 } 315 } 316 317 #[test] 318 fn dropping_a_polled_subscription_future_requests_cancellation() { 319 let targets = TargetSet::new(vec![target("wss://one.example")]).expect("targets"); 320 let request = request(targets, 1); 321 let terminal = SubscriptionEnd::for_request(&request, 0, [], SubscriptionEndReason::Cancelled) 322 .expect("terminal"); 323 let cancellation_observed = Arc::new(AtomicBool::new(false)); 324 let mut subscription = PendingSubscription { 325 request, 326 terminal, 327 cancellation_observed: Arc::clone(&cancellation_observed), 328 }; 329 330 let unpolled = subscription.next(); 331 drop(unpolled); 332 assert!(!cancellation_observed.load(Ordering::SeqCst)); 333 334 let mut pending = subscription.next(); 335 let mut context = Context::from_waker(noop_waker_ref()); 336 assert!(Pin::new(&mut pending).poll(&mut context).is_pending()); 337 drop(pending); 338 assert!(cancellation_observed.load(Ordering::SeqCst)); 339 assert_eq!( 340 block_on(subscription.cancel()).expect("cancel").reason(), 341 SubscriptionEndReason::Cancelled 342 ); 343 } 344 345 #[cfg(feature = "serde")] 346 #[test] 347 fn subscription_wire_models_revalidate_bounds_and_request_binding() { 348 let requested = target("wss://one.example"); 349 let request = request(TargetSet::new(vec![requested.clone()]).expect("targets"), 1); 350 let cursor = FetchCursor::parse("event-1").expect("cursor"); 351 let event = SubscriptionEvent::for_request( 352 &request, 353 ObservedEvent::new( 354 signed_event(), 355 EventProvenance::new(TransportId::NOSTR, requested.fingerprint().clone(), 1) 356 .expect("provenance") 357 .with_cursor(cursor.clone()), 358 ), 359 SubscriptionCheckpoint::new(requested.fingerprint().clone(), cursor), 360 ) 361 .expect("event"); 362 let encoded = serde_json::to_string(&SubscriptionNext::Event(Box::new(event.clone()))) 363 .expect("serialize event"); 364 assert_eq!( 365 serde_json::from_str::<SubscriptionNext>(&encoded).expect("deserialize event"), 366 SubscriptionNext::Event(Box::new(event)) 367 ); 368 369 let mut invalid = serde_json::to_value(&request).expect("request value"); 370 invalid["bounds"]["event_limit"] = 0.into(); 371 assert!(serde_json::from_value::<SubscriptionRequest>(invalid).is_err()); 372 373 let checkpoint = serde_json::to_value(SubscriptionCheckpoint::new( 374 requested.fingerprint().clone(), 375 FetchCursor::parse("checkpoint").expect("cursor"), 376 )) 377 .expect("checkpoint value"); 378 let mut oversized = serde_json::to_value(&request).expect("request value"); 379 oversized["checkpoints"] = serde_json::Value::Array(vec![ 380 checkpoint; 381 radroots_transport::TARGET_SET_MAX_ITEMS 382 + 1 383 ]); 384 assert!(serde_json::from_value::<SubscriptionRequest>(oversized).is_err()); 385 386 let terminal = SubscriptionEnd::for_request(&request, 1, [], SubscriptionEndReason::Deadline) 387 .expect("terminal"); 388 let mut unknown = serde_json::to_value(&terminal).expect("terminal value"); 389 unknown["unknown"] = true.into(); 390 assert!(serde_json::from_value::<SubscriptionEnd>(unknown).is_err()); 391 }