subscription.rs (21395B)
1 //! Bounded, independent host subscriptions for focused runtime invalidation signals. 2 3 use std::collections::BTreeMap; 4 use std::panic::{AssertUnwindSafe, catch_unwind}; 5 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering}; 6 use std::sync::{Arc, Mutex, Weak}; 7 8 use tera_core::runtime::invalidation::{InvalidationDomain, RuntimeInvalidations}; 9 use tera_core::runtime::product_surface::LocalNetwork; 10 11 use crate::subscription_queue::SubscriptionQueue; 12 use crate::{FfiRuntimeChangeDelivery, FfiRuntimeChangeKind, FfiRuntimeChangeRecord, TeraAppError}; 13 14 const MAX_SUBSCRIPTIONS: usize = 32; 15 16 #[uniffi::export(callback_interface)] 17 pub trait TeraRuntimeObserver: Send + Sync { 18 fn on_change(&self, change: FfiRuntimeChangeRecord); 19 } 20 21 pub(crate) struct SubscriptionHub { 22 next_id: AtomicU64, 23 source: RuntimeInvalidations, 24 closed: AtomicBool, 25 workers: Arc<WorkerState>, 26 subscriptions: Mutex<BTreeMap<u64, Arc<SubscriptionQueue>>>, 27 } 28 29 impl SubscriptionHub { 30 pub(crate) fn new(source: RuntimeInvalidations) -> Arc<Self> { 31 Arc::new(Self { 32 next_id: AtomicU64::new(1), 33 source, 34 closed: AtomicBool::new(false), 35 workers: Arc::new(WorkerState::default()), 36 subscriptions: Mutex::new(BTreeMap::new()), 37 }) 38 } 39 40 pub(crate) fn subscribe( 41 self: &Arc<Self>, 42 observer: Box<dyn TeraRuntimeObserver>, 43 ) -> Result<Arc<FfiSubscriptionHandle>, TeraAppError> { 44 let id = self 45 .next_id 46 .fetch_update(Ordering::AcqRel, Ordering::Acquire, |id| id.checked_add(1)) 47 .map_err(|_| subscription_error("subscription_limit_reached", false))?; 48 { 49 let mut subscriptions = self 50 .subscriptions 51 .lock() 52 .unwrap_or_else(std::sync::PoisonError::into_inner); 53 if self.closed.load(Ordering::Acquire) { 54 return Err(subscription_error("runtime_closed", false)); 55 } 56 if subscriptions.len() >= MAX_SUBSCRIPTIONS 57 || self.workers.active.load(Ordering::Acquire) >= MAX_SUBSCRIPTIONS 58 { 59 return Err(subscription_error("subscription_limit_reached", true)); 60 } 61 // Enqueue the initial snapshot before exposing this queue to any 62 // publisher. An observer always sees the epoch before later hints. 63 let queue = SubscriptionQueue::new( 64 self.source 65 .snapshot(InvalidationDomain::Initial, None) 66 .into(), 67 ); 68 let receiver = Arc::clone(&queue); 69 self.workers.active.fetch_add(1, Ordering::AcqRel); 70 let worker = WorkerLease(Arc::clone(&self.workers)); 71 let hub = Arc::downgrade(self); 72 std::thread::Builder::new() 73 .name(format!("tera-ffi-observer-{id}")) 74 .spawn(move || { 75 let _worker = worker; 76 while let Some(change) = receiver.receive() { 77 let Some(hub) = hub.upgrade() else { 78 break; 79 }; 80 let closed = hub.closed.load(Ordering::Acquire); 81 drop(hub); 82 if closed 83 && change.kind != FfiRuntimeChangeKind::Lifecycle 84 && change.delivery != FfiRuntimeChangeDelivery::ResnapshotRequired 85 { 86 continue; 87 } 88 if catch_unwind(AssertUnwindSafe(|| observer.on_change(change))).is_err() { 89 break; 90 } 91 } 92 if let Some(hub) = hub.upgrade() { 93 hub.remove(id); 94 } 95 }) 96 .map_err(|_| subscription_error("subscription_worker_unavailable", true))?; 97 subscriptions.insert(id, queue); 98 } 99 100 Ok(Arc::new(FfiSubscriptionHandle { 101 hub: Arc::downgrade(self), 102 id: Mutex::new(Some(id)), 103 })) 104 } 105 106 pub(crate) fn notify(&self, kind: FfiRuntimeChangeKind, entity_id: Option<String>) { 107 self.notify_context(kind, None, entity_id); 108 } 109 110 pub(crate) fn notify_context( 111 &self, 112 kind: FfiRuntimeChangeKind, 113 context: Option<&LocalNetwork>, 114 entity_id: Option<String>, 115 ) { 116 // Serialize revision assignment with nonblocking enqueue so concurrent 117 // publishers cannot deliver an older domain revision after a newer one. 118 let mut subscriptions = self 119 .subscriptions 120 .lock() 121 .unwrap_or_else(std::sync::PoisonError::into_inner); 122 if self.closed.load(Ordering::Acquire) { 123 return; 124 } 125 let change: FfiRuntimeChangeRecord = 126 self.source.advance(kind.into(), context, entity_id).into(); 127 subscriptions.retain(|_, queue| queue.send(change.clone())); 128 } 129 130 pub(crate) fn close(&self) { 131 if !self.closed.swap(true, Ordering::AcqRel) { 132 let mut subscriptions = self 133 .subscriptions 134 .lock() 135 .unwrap_or_else(std::sync::PoisonError::into_inner); 136 let change: FfiRuntimeChangeRecord = self 137 .source 138 .advance(InvalidationDomain::Lifecycle, None, None) 139 .into(); 140 for queue in subscriptions.values() { 141 queue.close(change.clone()); 142 } 143 subscriptions.clear(); 144 } 145 } 146 147 pub(crate) async fn drain(&self) { 148 loop { 149 let notified = self.workers.drained.notified(); 150 let mut notified = std::pin::pin!(notified); 151 notified.as_mut().enable(); 152 if self.workers.active.load(Ordering::Acquire) == 0 { 153 return; 154 } 155 notified.await; 156 } 157 } 158 159 fn remove(&self, id: u64) { 160 let removed = self 161 .subscriptions 162 .lock() 163 .unwrap_or_else(std::sync::PoisonError::into_inner) 164 .remove(&id); 165 if let Some(queue) = removed { 166 queue.cancel(); 167 } 168 } 169 } 170 171 impl Drop for SubscriptionHub { 172 fn drop(&mut self) { 173 self.close(); 174 } 175 } 176 177 // Workers retain only their settlement counter, avoiding a hub/sender cycle. 178 // Close rejects new observers and drains callbacks without blocking an executor. 179 #[derive(Default)] 180 struct WorkerState { 181 active: AtomicUsize, 182 drained: tokio::sync::Notify, 183 } 184 185 struct WorkerLease(Arc<WorkerState>); 186 187 impl Drop for WorkerLease { 188 fn drop(&mut self) { 189 if self.0.active.fetch_sub(1, Ordering::AcqRel) == 1 { 190 self.0.drained.notify_waiters(); 191 } 192 } 193 } 194 195 #[derive(uniffi::Object)] 196 pub struct FfiSubscriptionHandle { 197 hub: Weak<SubscriptionHub>, 198 id: Mutex<Option<u64>>, 199 } 200 201 #[uniffi::export] 202 impl FfiSubscriptionHandle { 203 pub fn unsubscribe(&self) { 204 let id = self 205 .id 206 .lock() 207 .unwrap_or_else(std::sync::PoisonError::into_inner) 208 .take(); 209 if let (Some(hub), Some(id)) = (self.hub.upgrade(), id) { 210 hub.remove(id); 211 } 212 } 213 214 pub fn is_active(&self) -> bool { 215 let id = *self 216 .id 217 .lock() 218 .unwrap_or_else(std::sync::PoisonError::into_inner); 219 let (Some(hub), Some(id)) = (self.hub.upgrade(), id) else { 220 return false; 221 }; 222 !hub.closed.load(Ordering::Acquire) 223 && hub 224 .subscriptions 225 .lock() 226 .unwrap_or_else(std::sync::PoisonError::into_inner) 227 .contains_key(&id) 228 } 229 } 230 231 impl Drop for FfiSubscriptionHandle { 232 fn drop(&mut self) { 233 let id = self 234 .id 235 .get_mut() 236 .unwrap_or_else(std::sync::PoisonError::into_inner) 237 .take(); 238 if let (Some(hub), Some(id)) = (self.hub.upgrade(), id) { 239 hub.remove(id); 240 } 241 } 242 } 243 244 fn subscription_error(code: &str, retryable: bool) -> TeraAppError { 245 TeraAppError::failure( 246 code, 247 "subscription", 248 retryable, 249 if retryable { &["retry"] } else { &[] }, 250 "The runtime change subscription is unavailable.", 251 ) 252 } 253 254 #[cfg(test)] 255 mod tests { 256 use std::sync::{Arc, Condvar}; 257 use std::time::{Duration, Instant}; 258 259 use super::*; 260 use crate::subscription_queue::CHANGE_BUFFER_CAPACITY; 261 262 fn test_hub() -> Arc<SubscriptionHub> { 263 let store = tera_core::runtime::store::MobileUserStoreConfig::from_encoded( 264 "/tmp/tera-invalidation-fixture", 265 "79be667ef9dcbbac55a06295ce870b07029bfcdb2dce28d959f2815b16f81798", 266 &"04".repeat(32), 267 1, 268 tera_core::runtime::store::ProtectedDataAvailability::Available, 269 ) 270 .unwrap(); 271 SubscriptionHub::new(RuntimeInvalidations::new( 272 store.public_key(), 273 store.source_generation(), 274 std::num::NonZeroU128::new(1).unwrap(), 275 )) 276 } 277 278 struct NoopObserver; 279 280 struct RecordingObserver(std::sync::mpsc::Sender<FfiRuntimeChangeRecord>); 281 282 impl TeraRuntimeObserver for RecordingObserver { 283 fn on_change(&self, change: FfiRuntimeChangeRecord) { 284 self.0.send(change).unwrap(); 285 } 286 } 287 288 #[test] 289 fn concurrent_publication_is_ordered_and_subscription_does_not_advance_domains() { 290 let hub = test_hub(); 291 let (sender, receiver) = std::sync::mpsc::channel(); 292 let first = hub.subscribe(Box::new(RecordingObserver(sender))).unwrap(); 293 let initial = receiver.recv_timeout(Duration::from_secs(5)).unwrap(); 294 assert_eq!(initial.kind, FfiRuntimeChangeKind::Initial); 295 assert_eq!(initial.schema_version, crate::RUNTIME_CHANGE_SCHEMA_VERSION); 296 let publishers = (0..8) 297 .map(|index| { 298 let hub = Arc::clone(&hub); 299 std::thread::spawn(move || { 300 hub.notify(FfiRuntimeChangeKind::Drafts, Some(index.to_string())) 301 }) 302 }) 303 .collect::<Vec<_>>(); 304 for publisher in publishers { 305 publisher.join().unwrap(); 306 } 307 for expected in 1..=8 { 308 let change = receiver.recv_timeout(Duration::from_secs(5)).unwrap(); 309 assert_eq!(change.epoch, initial.epoch); 310 assert_eq!(change.scope, initial.scope); 311 assert_eq!( 312 change.revision, 313 crate::FfiInvalidationRevision::Current { value: expected } 314 ); 315 } 316 first.unsubscribe(); 317 let before = hub.source.snapshot(InvalidationDomain::Drafts, None); 318 let (sender, receiver) = std::sync::mpsc::channel(); 319 let second = hub.subscribe(Box::new(RecordingObserver(sender))).unwrap(); 320 let resumed = receiver.recv_timeout(Duration::from_secs(5)).unwrap(); 321 assert_eq!(resumed.epoch, initial.epoch); 322 assert_eq!( 323 before, 324 hub.source.snapshot(InvalidationDomain::Drafts, None) 325 ); 326 second.unsubscribe(); 327 } 328 329 #[test] 330 fn subscription_identity_exhaustion_has_no_worker_or_revision_side_effect() { 331 let hub = test_hub(); 332 hub.next_id.store(u64::MAX, Ordering::Release); 333 let before = hub.source.snapshot(InvalidationDomain::Initial, None); 334 let failure = hub.subscribe(Box::new(NoopObserver)).err().unwrap(); 335 assert_eq!(failure.report().code, "subscription_limit_reached"); 336 assert!(!failure.report().retryable); 337 assert_eq!(hub.next_id.load(Ordering::Acquire), u64::MAX); 338 assert_eq!(hub.workers.active.load(Ordering::Acquire), 0); 339 assert_eq!( 340 before, 341 hub.source.snapshot(InvalidationDomain::Initial, None) 342 ); 343 } 344 345 impl TeraRuntimeObserver for NoopObserver { 346 fn on_change(&self, _change: FfiRuntimeChangeRecord) {} 347 } 348 349 struct PanicObserver; 350 351 impl TeraRuntimeObserver for PanicObserver { 352 fn on_change(&self, _change: FfiRuntimeChangeRecord) { 353 panic!("observer panic is isolated"); 354 } 355 } 356 357 struct BlockingObserver(Arc<(Mutex<bool>, Condvar)>); 358 359 impl TeraRuntimeObserver for BlockingObserver { 360 fn on_change(&self, change: FfiRuntimeChangeRecord) { 361 if change.kind == FfiRuntimeChangeKind::Initial { 362 let (released, wake) = &*self.0; 363 let guard = released 364 .lock() 365 .unwrap_or_else(std::sync::PoisonError::into_inner); 366 let _guard = wake 367 .wait_while(guard, |released| !*released) 368 .unwrap_or_else(std::sync::PoisonError::into_inner); 369 } 370 } 371 } 372 373 struct PausedObserver { 374 entered: Arc<tokio::sync::Notify>, 375 gate: Arc<(Mutex<bool>, Condvar)>, 376 calls: Arc<AtomicUsize>, 377 } 378 379 impl TeraRuntimeObserver for PausedObserver { 380 fn on_change(&self, _: FfiRuntimeChangeRecord) { 381 self.calls.fetch_add(1, Ordering::AcqRel); 382 self.entered.notify_one(); 383 let (released, wake) = &*self.gate; 384 drop( 385 wake.wait_while(released.lock().unwrap(), |released| !*released) 386 .unwrap(), 387 ); 388 } 389 } 390 391 #[tokio::test] 392 async fn close_drains_native_callbacks_even_after_the_wait_is_cancelled() { 393 use std::{ 394 future::Future, 395 task::{Context, Poll, Waker}, 396 }; 397 let hub = test_hub(); 398 let entered = Arc::new(tokio::sync::Notify::new()); 399 let gate = Arc::new((Mutex::new(false), Condvar::new())); 400 let calls = Arc::new(AtomicUsize::new(0)); 401 let handle = hub 402 .subscribe(Box::new(PausedObserver { 403 entered: Arc::clone(&entered), 404 gate: Arc::clone(&gate), 405 calls: Arc::clone(&calls), 406 })) 407 .unwrap(); 408 tokio::time::timeout(Duration::from_secs(5), entered.notified()) 409 .await 410 .unwrap(); 411 hub.notify(FfiRuntimeChangeKind::Today, None); 412 hub.close(); 413 let mut drain = Box::pin(hub.drain()); 414 assert!(matches!( 415 drain.as_mut().poll(&mut Context::from_waker(Waker::noop())), 416 Poll::Pending 417 )); 418 assert!(!handle.is_active()); 419 assert_eq!( 420 hub.subscribe(Box::new(NoopObserver)) 421 .err() 422 .unwrap() 423 .report() 424 .code, 425 "runtime_closed" 426 ); 427 drop(drain); 428 *gate.0.lock().unwrap() = true; 429 gate.1.notify_all(); 430 tokio::time::timeout(Duration::from_secs(5), hub.drain()) 431 .await 432 .unwrap(); 433 hub.close(); 434 hub.drain().await; 435 assert_eq!(calls.load(Ordering::Acquire), 2); 436 assert_eq!(hub.workers.active.load(Ordering::Acquire), 0); 437 } 438 439 #[tokio::test] 440 async fn unsubscribe_discards_queued_callbacks_and_drains_the_admitted_callback() { 441 let hub = test_hub(); 442 let entered = Arc::new(tokio::sync::Notify::new()); 443 let gate = Arc::new((Mutex::new(false), Condvar::new())); 444 let calls = Arc::new(AtomicUsize::new(0)); 445 let handle = hub 446 .subscribe(Box::new(PausedObserver { 447 entered: Arc::clone(&entered), 448 gate: Arc::clone(&gate), 449 calls: Arc::clone(&calls), 450 })) 451 .unwrap(); 452 tokio::time::timeout(Duration::from_secs(5), entered.notified()) 453 .await 454 .unwrap(); 455 for _ in 0..=CHANGE_BUFFER_CAPACITY { 456 hub.notify(FfiRuntimeChangeKind::Today, None); 457 } 458 handle.unsubscribe(); 459 assert!(!handle.is_active()); 460 assert_eq!(hub.workers.active.load(Ordering::Acquire), 1); 461 *gate.0.lock().unwrap() = true; 462 gate.1.notify_all(); 463 tokio::time::timeout(Duration::from_secs(5), hub.drain()) 464 .await 465 .unwrap(); 466 assert_eq!(calls.load(Ordering::Acquire), 1); 467 assert_eq!(hub.workers.active.load(Ordering::Acquire), 0); 468 } 469 470 #[test] 471 fn dropping_the_hub_releases_an_idle_observer_worker() { 472 let hub = test_hub(); 473 let workers = Arc::clone(&hub.workers); 474 let (sender, receiver) = std::sync::mpsc::channel(); 475 let handle = hub.subscribe(Box::new(RecordingObserver(sender))).unwrap(); 476 receiver.recv_timeout(Duration::from_secs(5)).unwrap(); 477 drop(hub); 478 assert!(!handle.is_active()); 479 let deadline = Instant::now() + Duration::from_secs(5); 480 while workers.active.load(Ordering::Acquire) != 0 && Instant::now() < deadline { 481 std::thread::yield_now(); 482 } 483 assert_eq!(workers.active.load(Ordering::Acquire), 0); 484 } 485 486 #[tokio::test] 487 async fn cancelled_callback_keeps_worker_admission_until_it_actually_returns() { 488 let hub = test_hub(); 489 let entered = Arc::new(tokio::sync::Notify::new()); 490 let gate = Arc::new((Mutex::new(false), Condvar::new())); 491 let paused = hub 492 .subscribe(Box::new(PausedObserver { 493 entered: Arc::clone(&entered), 494 gate: Arc::clone(&gate), 495 calls: Arc::new(AtomicUsize::new(0)), 496 })) 497 .unwrap(); 498 tokio::time::timeout(Duration::from_secs(5), entered.notified()) 499 .await 500 .unwrap(); 501 let others = (1..MAX_SUBSCRIPTIONS) 502 .map(|_| hub.subscribe(Box::new(NoopObserver)).unwrap()) 503 .collect::<Vec<_>>(); 504 paused.unsubscribe(); 505 let attempted = hub.subscribe(Box::new(NoopObserver)); 506 let active = hub.workers.active.load(Ordering::Acquire); 507 // Release even if the assertion rejects an over-admitted mutant. 508 *gate.0.lock().unwrap() = true; 509 gate.1.notify_all(); 510 assert_eq!(active, MAX_SUBSCRIPTIONS); 511 let error = attempted 512 .err() 513 .expect("cancelled callback still owns its worker slot"); 514 assert_eq!(error.report().code, "subscription_limit_reached"); 515 assert!(error.report().retryable); 516 tokio::time::timeout(Duration::from_secs(5), async { 517 while hub.workers.active.load(Ordering::Acquire) == MAX_SUBSCRIPTIONS { 518 tokio::task::yield_now().await; 519 } 520 }) 521 .await 522 .unwrap(); 523 let resumed = hub.subscribe(Box::new(NoopObserver)).unwrap(); 524 drop(resumed); 525 drop(others); 526 hub.close(); 527 tokio::time::timeout(Duration::from_secs(5), hub.drain()) 528 .await 529 .unwrap(); 530 assert_eq!(hub.workers.active.load(Ordering::Acquire), 0); 531 } 532 533 #[test] 534 fn closed_limit_and_detached_handle_paths_are_typed_and_idempotent() { 535 let closed = test_hub(); 536 closed.close(); 537 closed.close(); 538 closed.notify(FfiRuntimeChangeKind::Today, None); 539 let error = closed 540 .subscribe(Box::new(NoopObserver)) 541 .err() 542 .expect("closed hub"); 543 assert_eq!(error.report().code, "runtime_closed"); 544 assert!(!error.report().retryable); 545 546 let hub = test_hub(); 547 let handles = (0..MAX_SUBSCRIPTIONS) 548 .map(|_| hub.subscribe(Box::new(NoopObserver)).expect("subscription")) 549 .collect::<Vec<_>>(); 550 let error = hub 551 .subscribe(Box::new(NoopObserver)) 552 .err() 553 .expect("bounded subscription limit"); 554 assert_eq!(error.report().code, "subscription_limit_reached"); 555 assert!(error.report().retryable); 556 drop(handles); 557 558 let detached_hub = test_hub(); 559 let detached = detached_hub 560 .subscribe(Box::new(NoopObserver)) 561 .expect("detached subscription"); 562 drop(detached_hub); 563 assert!(!detached.is_active()); 564 detached.unsubscribe(); 565 detached.unsubscribe(); 566 } 567 568 #[test] 569 fn callback_panics_and_full_buffers_never_escape_or_block_publishers() { 570 let hub = test_hub(); 571 let panicking = hub 572 .subscribe(Box::new(PanicObserver)) 573 .expect("panicking subscription"); 574 let deadline = Instant::now() + Duration::from_secs(1); 575 while panicking.is_active() && Instant::now() < deadline { 576 std::thread::yield_now(); 577 } 578 assert!(!panicking.is_active()); 579 580 let release = Arc::new((Mutex::new(false), Condvar::new())); 581 let blocked = hub 582 .subscribe(Box::new(BlockingObserver(Arc::clone(&release)))) 583 .expect("blocked subscription"); 584 for generation in 0..=CHANGE_BUFFER_CAPACITY { 585 hub.notify( 586 FfiRuntimeChangeKind::Drafts, 587 Some(format!("draft-{generation}")), 588 ); 589 } 590 assert!(blocked.is_active()); 591 let (released, wake) = &*release; 592 *released 593 .lock() 594 .unwrap_or_else(std::sync::PoisonError::into_inner) = true; 595 wake.notify_all(); 596 hub.close(); 597 assert!(!blocked.is_active()); 598 } 599 }