source_contract.rs (20003B)
1 use core::{future::Future, pin::Pin, task::Context}; 2 use std::sync::{ 3 Arc, 4 atomic::{AtomicBool, Ordering}, 5 }; 6 7 use futures::{future, task::noop_waker_ref}; 8 use radroots_event::{SignedEvent, wire::v1::Nip01EventWire}; 9 use radroots_identity::PublicKey; 10 use radroots_transport::{ 11 BoxFuture, Error, EventSource, FetchPage, FetchRequest, SourceStatus, Target, TargetSet, 12 TransportId, 13 capability::{Availability, Maturity, SourceCapabilities}, 14 outcome::{FetchTargetOutcome, FetchTargetState}, 15 source::{ 16 EventProvenance, FETCH_CURSOR_MAX_BYTES, FETCH_PAGE_MAX_EVENTS, FETCH_REQUEST_ID_MAX_BYTES, 17 FETCH_SELECTOR_MAX_AUTHORS, FETCH_SELECTOR_MAX_KINDS, FETCH_SELECTOR_MAX_TAG_VALUES, 18 FETCH_SELECTOR_TAG_VALUE_MAX_BYTES, FetchBounds, FetchCursor, FetchSelector, NextPage, 19 ObservedEvent, 20 }, 21 }; 22 23 fn target(uri: &str) -> Target { 24 Target::nostr_relay(uri).expect("nostr target") 25 } 26 27 #[test] 28 fn fetch_selector_is_bounded_canonical_and_request_bound() { 29 let event = tagged_event(); 30 let author = *event.pubkey(); 31 let selector = FetchSelector::all() 32 .with_kinds(vec![1, 0]) 33 .expect("kind selector") 34 .with_authors(vec![author]) 35 .expect("author selector") 36 .with_exact_tag_value('d', "trade-1") 37 .expect("tag selector") 38 .with_since_unix_seconds(1_700_000_000) 39 .expect("since") 40 .with_until_unix_seconds(1_700_000_100) 41 .expect("until"); 42 43 assert_eq!(selector.kinds(), &[0, 1]); 44 assert_eq!(selector.authors(), &[author]); 45 let exact_tags = selector.exact_tag_filters().collect::<Vec<_>>(); 46 assert_eq!(exact_tags.len(), 1); 47 assert_eq!(exact_tags[0].0, 'd'); 48 assert_eq!(exact_tags[0].1, &[String::from("trade-1")]); 49 assert!(selector.matches(&event)); 50 #[cfg(feature = "serde")] 51 { 52 let encoded = serde_json::to_string(&selector).expect("selector JSON"); 53 assert_eq!( 54 serde_json::from_str::<FetchSelector>(encoded.as_str()).expect("selector round trip"), 55 selector 56 ); 57 assert!( 58 serde_json::from_value::<FetchSelector>(serde_json::json!({ 59 "kinds": [], 60 "authors": [], 61 "exact_tags": {"D": ["trade-1"]}, 62 "since_unix_seconds": null, 63 "until_unix_seconds": null 64 })) 65 .is_err() 66 ); 67 } 68 assert_eq!( 69 FetchSelector::all() 70 .with_kinds(vec![1, 1]) 71 .expect_err("duplicate kind"), 72 Error::DuplicateFetchKind 73 ); 74 assert_eq!( 75 FetchSelector::all() 76 .with_authors(vec![author, author]) 77 .expect_err("duplicate author"), 78 Error::DuplicateFetchAuthor 79 ); 80 assert_eq!( 81 FetchSelector::all() 82 .with_kinds(vec![0; FETCH_SELECTOR_MAX_KINDS + 1]) 83 .expect_err("too many kinds"), 84 Error::FetchSelectorTooLarge 85 ); 86 assert_eq!( 87 FetchSelector::all() 88 .with_authors(vec![author; FETCH_SELECTOR_MAX_AUTHORS + 1]) 89 .expect_err("too many authors"), 90 Error::FetchSelectorTooLarge 91 ); 92 for invalid in ['D', '0', '#', 'é'] { 93 assert_eq!( 94 FetchSelector::all() 95 .with_exact_tag_value(invalid, "trade-1") 96 .expect_err("invalid tag key"), 97 Error::InvalidFetchTagKey 98 ); 99 } 100 for invalid in [String::new(), String::from("line\nbreak")] { 101 assert_eq!( 102 FetchSelector::all() 103 .with_exact_tag_value('d', invalid) 104 .expect_err("invalid tag value"), 105 Error::InvalidFetchTagValue 106 ); 107 } 108 assert_eq!( 109 FetchSelector::all() 110 .with_exact_tag_value('d', "x".repeat(FETCH_SELECTOR_TAG_VALUE_MAX_BYTES + 1)) 111 .expect_err("oversized tag value"), 112 Error::InvalidFetchTagValue 113 ); 114 assert!( 115 FetchSelector::all() 116 .with_exact_tag_value('d', "x".repeat(FETCH_SELECTOR_TAG_VALUE_MAX_BYTES)) 117 .is_ok() 118 ); 119 assert_eq!( 120 FetchSelector::all() 121 .with_exact_tag_value('d', "trade-1") 122 .and_then(|selector| selector.with_exact_tag_value('d', "trade-1")) 123 .expect_err("duplicate tag value"), 124 Error::DuplicateFetchTagValue 125 ); 126 let maximum = (0..FETCH_SELECTOR_MAX_TAG_VALUES) 127 .try_fold(FetchSelector::all(), |selector, index| { 128 selector.with_exact_tag_value('d', format!("trade-{index:03}")) 129 }); 130 assert!(maximum.is_ok()); 131 assert_eq!( 132 maximum 133 .and_then(|selector| selector.with_exact_tag_value('d', "trade-overflow")) 134 .expect_err("too many tag values"), 135 Error::FetchSelectorTooLarge 136 ); 137 let every_key = (0..radroots_transport::source::FETCH_SELECTOR_MAX_TAG_KEYS).try_fold( 138 FetchSelector::all(), 139 |selector, index| { 140 selector.with_exact_tag_value( 141 char::from(b'a' + u8::try_from(index).expect("bounded key index")), 142 "value", 143 ) 144 }, 145 ); 146 assert_eq!( 147 every_key 148 .expect("all lowercase keys") 149 .exact_tag_filters() 150 .count(), 151 radroots_transport::source::FETCH_SELECTOR_MAX_TAG_KEYS 152 ); 153 #[cfg(feature = "serde")] 154 assert!( 155 serde_json::from_str::<FetchSelector>( 156 r#"{"kinds":[],"authors":[],"exact_tags":{"d":["one"],"d":["two"]},"since_unix_seconds":null,"until_unix_seconds":null}"#, 157 ) 158 .is_err() 159 ); 160 assert_eq!( 161 FetchSelector::all() 162 .with_since_unix_seconds(2) 163 .and_then(|selector| selector.with_until_unix_seconds(1)) 164 .expect_err("reversed range"), 165 Error::InvalidFetchTimeRange 166 ); 167 168 let targets = TargetSet::new(vec![target("wss://one.example")]).expect("targets"); 169 let selected = request(targets.clone(), 1).with_selector(selector); 170 let page = FetchPage::for_request(&selected, Vec::new(), Vec::new(), NextPage::Complete) 171 .expect("selected page"); 172 assert_eq!( 173 page.validate_for_request(&request(targets, 1)) 174 .expect_err("selector mismatch"), 175 Error::FetchPageRequestMismatch 176 ); 177 } 178 179 fn signed_event() -> SignedEvent { 180 let raw = r#"{"id":"56bfc78223bb2221bad82b539efdec1ade0f56d0eb0e1f592fd387df4b2ceee0","pubkey":"585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df","created_at":1700000001,"kind":0,"tags":[],"content":"{}","sig":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"}"#; 181 let wire = Nip01EventWire::parse_json(raw).expect("wire event"); 182 SignedEvent::from_wire_verified_id(wire, raw).expect("signed event") 183 } 184 185 fn tagged_event() -> SignedEvent { 186 let raw = r#"{"id":"56bfc78223bb2221bad82b539efdec1ade0f56d0eb0e1f592fd387df4b2ceee0","pubkey":"585591529da0bab31b3b1b1f986611cf5f435dca84f978c89ee8a40cca7103df","created_at":1700000001,"kind":0,"tags":[],"content":"{}","sig":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"}"#; 187 let mut wire = Nip01EventWire::parse_json(raw).expect("wire event"); 188 wire.tags = vec![vec![String::from("d"), String::from("trade-1")]]; 189 wire.id = wire.computed_event_id().expect("event id").into_string(); 190 let mut value: serde_json::Value = serde_json::from_str(raw).expect("fixture JSON"); 191 value["id"] = serde_json::json!(&wire.id); 192 value["tags"] = serde_json::json!(&wire.tags); 193 let raw = value.to_string(); 194 SignedEvent::from_wire_verified_id(wire, raw.as_str()).expect("signed event") 195 } 196 197 fn request(targets: TargetSet, limit: u16) -> FetchRequest { 198 FetchRequest::new( 199 "fetch-request", 200 targets, 201 FetchBounds::new(limit, 1_700_000_100_000).expect("bounds"), 202 ) 203 .expect("request") 204 } 205 206 #[test] 207 fn fetch_bounds_request_ids_and_cursors_fail_closed() { 208 assert_eq!( 209 FetchBounds::new(0, 1).expect_err("zero limit"), 210 Error::InvalidFetchLimit 211 ); 212 assert_eq!( 213 FetchBounds::new(FETCH_PAGE_MAX_EVENTS + 1, 1).expect_err("oversized limit"), 214 Error::InvalidFetchLimit 215 ); 216 assert_eq!( 217 FetchBounds::new(1, 0).expect_err("zero deadline"), 218 Error::InvalidFetchDeadline 219 ); 220 assert_eq!( 221 FetchRequest::new( 222 "", 223 TargetSet::new(vec![target("wss://one.example")]).expect("targets"), 224 FetchBounds::new(1, 1).expect("bounds"), 225 ) 226 .expect_err("empty request id"), 227 Error::EmptyFetchRequestId 228 ); 229 assert_eq!( 230 FetchRequest::new( 231 "x".repeat(FETCH_REQUEST_ID_MAX_BYTES + 1), 232 TargetSet::new(vec![target("wss://one.example")]).expect("targets"), 233 FetchBounds::new(1, 1).expect("bounds"), 234 ) 235 .expect_err("oversized request id"), 236 Error::InvalidFetchRequestId 237 ); 238 assert_eq!( 239 FetchCursor::parse("").expect_err("empty cursor"), 240 Error::EmptyFetchCursor 241 ); 242 assert_eq!( 243 FetchCursor::parse("x".repeat(FETCH_CURSOR_MAX_BYTES + 1)).expect_err("oversized cursor"), 244 Error::InvalidFetchCursor 245 ); 246 for invalid in [" request", "request ", "request\nid"] { 247 assert_eq!( 248 FetchRequest::new( 249 invalid, 250 TargetSet::new(vec![target("wss://one.example")]).expect("targets"), 251 FetchBounds::new(1, 1).expect("bounds"), 252 ) 253 .expect_err("invalid request id"), 254 Error::InvalidFetchRequestId 255 ); 256 } 257 for invalid in [" cursor", "cursor ", "cursor\nid"] { 258 assert_eq!( 259 FetchCursor::parse(invalid).expect_err("invalid cursor"), 260 Error::InvalidFetchCursor 261 ); 262 } 263 264 let bounds = FetchBounds::new(FETCH_PAGE_MAX_EVENTS, u64::MAX).expect("maximum bounds"); 265 assert_eq!(bounds.limit(), FETCH_PAGE_MAX_EVENTS); 266 assert_eq!(bounds.deadline_unix_ms(), u64::MAX); 267 let cursor = FetchCursor::parse("cursor").expect("cursor"); 268 assert_eq!(cursor.as_str(), "cursor"); 269 assert_eq!(cursor.to_string(), "cursor"); 270 } 271 272 #[test] 273 fn selectors_expose_bounds_and_reject_each_nonmatching_dimension() { 274 let event = signed_event(); 275 let other_author = 276 PublicKey::from_hex("79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798") 277 .expect("other author"); 278 279 let reversed_since = FetchSelector::all() 280 .with_until_unix_seconds(1) 281 .expect("until") 282 .with_since_unix_seconds(2) 283 .expect_err("reversed since"); 284 assert_eq!(reversed_since, Error::InvalidFetchTimeRange); 285 286 for selector in [ 287 FetchSelector::all().with_kinds(vec![1]).expect("kinds"), 288 FetchSelector::all() 289 .with_authors(vec![other_author]) 290 .expect("authors"), 291 FetchSelector::all() 292 .with_exact_tag_value('d', "other-trade") 293 .expect("tag"), 294 FetchSelector::all() 295 .with_since_unix_seconds(event.created_at() + 1) 296 .expect("since"), 297 FetchSelector::all() 298 .with_until_unix_seconds(event.created_at() - 1) 299 .expect("until"), 300 ] { 301 assert!(!selector.matches(&event)); 302 } 303 304 let selector = FetchSelector::all() 305 .with_since_unix_seconds(event.created_at()) 306 .expect("since") 307 .with_until_unix_seconds(event.created_at()) 308 .expect("until"); 309 assert_eq!(selector.since_unix_seconds(), Some(event.created_at())); 310 assert_eq!(selector.until_unix_seconds(), Some(event.created_at())); 311 assert!(selector.matches(&event)); 312 } 313 314 #[test] 315 fn page_preserves_cursor_provenance_and_partial_target_outcomes() { 316 let first = target("wss://one.example"); 317 let second = target("wss://two.example"); 318 let targets = TargetSet::new(vec![first.clone(), second.clone()]).expect("targets"); 319 let request = request(targets, 2).with_cursor(FetchCursor::parse("page-1").expect("cursor")); 320 let provenance = EventProvenance::new( 321 TransportId::NOSTR, 322 first.fingerprint().clone(), 323 1_700_000_000_001, 324 ) 325 .expect("provenance") 326 .with_cursor(FetchCursor::parse("event-1").expect("event cursor")); 327 let observed = ObservedEvent::new(signed_event(), provenance); 328 let outcomes = vec![ 329 FetchTargetOutcome::new(first.fingerprint().clone(), FetchTargetState::Complete), 330 FetchTargetOutcome::new( 331 second.fingerprint().clone(), 332 FetchTargetState::FailedRetryable, 333 ) 334 .with_message("relay unavailable"), 335 ]; 336 let page = FetchPage::for_request( 337 &request, 338 vec![observed], 339 outcomes, 340 NextPage::Cursor(FetchCursor::parse("page-2").expect("next cursor")), 341 ) 342 .expect("page"); 343 344 page.validate_for_request(&request) 345 .expect("request binding"); 346 assert_eq!(page.events()[0].event().id_str(), signed_event().id_str()); 347 assert_eq!(page.events()[0].provenance().target(), first.fingerprint()); 348 assert_eq!(page.target_outcomes().len(), 2); 349 assert!(page.target_outcomes()[1].state().is_retryable()); 350 assert_eq!( 351 page.target_outcomes()[1].message(), 352 Some("relay unavailable") 353 ); 354 assert!(matches!(page.next_page(), NextPage::Cursor(cursor) if cursor.as_str() == "page-2")); 355 356 #[cfg(feature = "serde")] 357 { 358 let encoded = serde_json::to_string(&page).expect("serialize page"); 359 assert!(!encoded.contains("admission")); 360 assert!(!encoded.contains("storage")); 361 assert_eq!( 362 serde_json::from_str::<FetchPage>(&encoded).expect("deserialize page"), 363 page 364 ); 365 let mut invalid_time = serde_json::to_value(&page).expect("page value"); 366 invalid_time["events"][0]["provenance"]["observed_at_unix_ms"] = 0.into(); 367 assert!(serde_json::from_value::<FetchPage>(invalid_time).is_err()); 368 } 369 } 370 371 #[test] 372 fn pages_reject_oversize_unrequested_and_duplicate_evidence() { 373 let requested = target("wss://one.example"); 374 let foreign = target("wss://foreign.example"); 375 let request = request(TargetSet::new(vec![requested.clone()]).expect("targets"), 1); 376 let observed = ObservedEvent::new( 377 signed_event(), 378 EventProvenance::new(TransportId::NOSTR, requested.fingerprint().clone(), 1) 379 .expect("provenance"), 380 ); 381 assert_eq!( 382 FetchPage::for_request( 383 &request, 384 vec![observed.clone(), observed], 385 Vec::new(), 386 NextPage::Complete, 387 ) 388 .expect_err("oversized page"), 389 Error::FetchPageLimitExceeded 390 ); 391 392 let foreign_observation = ObservedEvent::new( 393 signed_event(), 394 EventProvenance::new(TransportId::NOSTR, foreign.fingerprint().clone(), 1) 395 .expect("foreign provenance"), 396 ); 397 assert_eq!( 398 FetchPage::for_request( 399 &request, 400 vec![foreign_observation], 401 Vec::new(), 402 NextPage::Complete, 403 ) 404 .expect_err("foreign provenance"), 405 Error::UnexpectedFetchProvenance 406 ); 407 408 let wrong_transport = ObservedEvent::new( 409 signed_event(), 410 EventProvenance::new( 411 TransportId::parse("future-mesh").expect("custom transport"), 412 requested.fingerprint().clone(), 413 1, 414 ) 415 .expect("wrong transport provenance"), 416 ); 417 assert_eq!( 418 FetchPage::for_request( 419 &request, 420 vec![wrong_transport], 421 Vec::new(), 422 NextPage::Complete, 423 ) 424 .expect_err("transport mismatch"), 425 Error::UnexpectedFetchProvenance 426 ); 427 428 let duplicate = 429 FetchTargetOutcome::new(requested.fingerprint().clone(), FetchTargetState::Partial); 430 assert_eq!( 431 FetchPage::for_request( 432 &request, 433 Vec::new(), 434 vec![duplicate.clone(), duplicate], 435 NextPage::Cancelled { 436 resume_from: Some(FetchCursor::parse("resume").expect("resume cursor")), 437 }, 438 ) 439 .expect_err("duplicate outcome"), 440 Error::DuplicateFetchTargetOutcome 441 ); 442 assert_eq!( 443 FetchPage::for_request( 444 &request, 445 Vec::new(), 446 vec![FetchTargetOutcome::new( 447 foreign.fingerprint().clone(), 448 FetchTargetState::Unavailable, 449 )], 450 NextPage::Complete, 451 ) 452 .expect_err("foreign outcome"), 453 Error::UnexpectedFetchTargetOutcome 454 ); 455 456 let page = FetchPage::for_request(&request, Vec::new(), Vec::new(), NextPage::Complete) 457 .expect("empty page"); 458 let other_request = FetchRequest::new( 459 "other-request", 460 request.target_set().clone(), 461 request.bounds(), 462 ) 463 .expect("other request"); 464 assert_eq!( 465 page.validate_for_request(&other_request) 466 .expect_err("request mismatch"), 467 Error::FetchPageRequestMismatch 468 ); 469 470 let filtered = FetchRequest::new( 471 "filtered-request", 472 request.target_set().clone(), 473 request.bounds(), 474 ) 475 .expect("filtered request") 476 .with_selector( 477 FetchSelector::all() 478 .with_kinds(vec![1]) 479 .expect("filtered selector"), 480 ); 481 let unexpected = ObservedEvent::new( 482 signed_event(), 483 EventProvenance::new(TransportId::NOSTR, requested.fingerprint().clone(), 1) 484 .expect("provenance"), 485 ); 486 assert_eq!( 487 FetchPage::for_request(&filtered, vec![unexpected], Vec::new(), NextPage::Complete) 488 .expect_err("selector mismatch"), 489 Error::UnexpectedFetchEvent 490 ); 491 492 assert_eq!( 493 EventProvenance::new(TransportId::NOSTR, requested.fingerprint().clone(), 0) 494 .expect_err("zero observation time"), 495 Error::InvalidObservedAt 496 ); 497 } 498 499 struct CancellationSource { 500 published: Arc<AtomicBool>, 501 cancelled_after_publish: Arc<AtomicBool>, 502 } 503 504 struct PublicationGuard(Arc<AtomicBool>); 505 506 impl Drop for PublicationGuard { 507 fn drop(&mut self) { 508 self.0.store(true, Ordering::SeqCst); 509 } 510 } 511 512 impl EventSource for CancellationSource { 513 fn status(&self) -> BoxFuture<'_, Result<SourceStatus, Error>> { 514 Box::pin(async { 515 Ok(SourceStatus::new( 516 TransportId::NOSTR, 517 true, 518 Maturity::Stable, 519 Availability::Available, 520 SourceCapabilities::FETCH, 521 "ready", 522 )) 523 }) 524 } 525 526 fn fetch(&self, _request: FetchRequest) -> BoxFuture<'_, Result<FetchPage, Error>> { 527 let published = Arc::clone(&self.published); 528 let cancelled = Arc::clone(&self.cancelled_after_publish); 529 Box::pin(async move { 530 published.store(true, Ordering::SeqCst); 531 let _guard = PublicationGuard(cancelled); 532 future::pending::<Result<FetchPage, Error>>().await 533 }) 534 } 535 } 536 537 #[test] 538 fn dropping_fetch_futures_respects_before_and_after_publication_boundaries() { 539 let source = CancellationSource { 540 published: Arc::new(AtomicBool::new(false)), 541 cancelled_after_publish: Arc::new(AtomicBool::new(false)), 542 }; 543 let targets = TargetSet::new(vec![target("wss://one.example")]).expect("targets"); 544 545 let unpolled = source.fetch(request(targets.clone(), 1)); 546 drop(unpolled); 547 assert!(!source.published.load(Ordering::SeqCst)); 548 assert!(!source.cancelled_after_publish.load(Ordering::SeqCst)); 549 550 let mut published = source.fetch(request(targets, 1)); 551 let mut context = Context::from_waker(noop_waker_ref()); 552 assert!(Pin::new(&mut published).poll(&mut context).is_pending()); 553 assert!(source.published.load(Ordering::SeqCst)); 554 drop(published); 555 assert!(source.cancelled_after_publish.load(Ordering::SeqCst)); 556 }