Harden concurrency and resource lifecycle (#101)

crates/libremetaverse-voice-webrtc/src/native.rs

@@ -20,7 +20,7 @@ use std::io::Cursor;
 use std::net::{IpAddr, Ipv4Addr, SocketAddr};
 use std::path::Path;
 use std::pin::Pin;
-use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use str0m::change::{SdpAnswer, SdpOffer};
@@ -1426,6 +1426,7 @@ where
 /// Deterministic native WebRTC answerer used by CI and offline diagnostics.
 pub struct LoopbackSignaling {
     state: tokio::sync::Mutex<LoopbackState>,
+    provisioning: AtomicBool,
     active_tasks: Arc<AtomicUsize>,
     received_messages: Arc<Mutex<Vec<String>>>,
     peer_id: UUID,
@@ -1448,6 +1449,7 @@ impl LoopbackSignaling {
                 completed: false,
                 closed: false,
             }),
+            provisioning: AtomicBool::new(false),
             active_tasks: Arc::new(AtomicUsize::new(0)),
             received_messages: Arc::new(Mutex::new(Vec::new())),
             peer_id,
@@ -1481,10 +1483,21 @@ impl LoopbackSignaling {
 impl VoiceSignaling for LoopbackSignaling {
     fn provision(&self, request: ProvisionRequest) -> SignalFuture<'_, ProvisionResponse> {
         Box::pin(async move {
-            let mut state = self.state.lock().await;
-            if state.task.is_some() {
+            struct ProvisionGuard<'a>(&'a AtomicBool);
+            impl Drop for ProvisionGuard<'_> {
+                fn drop(&mut self) {
+                    self.0.store(false, Ordering::Release);
+                }
+            }
+
+            if self
+                .provisioning
+                .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
+                .is_err()
+            {
                 return Err(WebRtcError::Signaling("duplicate provision request".into()));
             }
+            let _provisioning = ProvisionGuard(&self.provisioning);
             if request.jsep.kind != "offer"
                 || request.channel_type != "local"
                 || request.voice_server_type != "webrtc"
@@ -1511,6 +1524,10 @@ impl VoiceSignaling for LoopbackSignaling {
             let viewer_session = UUID::random()
                 .map_err(|_| WebRtcError::Signaling("could not create session id".into()))?;
             let credentials = VoiceSecret::new("offline-secret")?;
+            let mut state = self.state.lock().await;
+            if state.task.is_some() {
+                return Err(WebRtcError::Signaling("duplicate provision request".into()));
+            }
             active.fetch_add(1, Ordering::AcqRel);
             let task = tokio::spawn(async move {
                 Box::pin(run_loopback_peer(
@@ -1815,6 +1832,33 @@ mod tests {
         assert_eq!(signaling.active_tasks(), 0);
     }
 
+    #[tokio::test]
+    async fn concurrent_provisioning_admits_one_session_without_leaking_the_loser() {
+        let peer = UUID::new_with_string("11111111-2222-4333-8444-555555555555".into()).unwrap();
+        let signaling = LoopbackSignaling::new(peer);
+        let config = VoiceSessionConfig {
+            timeout: Duration::from_secs(10),
+            ..VoiceSessionConfig::default()
+        };
+        let (first, second) = tokio::join!(
+            WebRtcVoiceSession::connect(signaling.clone(), config.clone()),
+            WebRtcVoiceSession::connect(signaling.clone(), config),
+        );
+        let (mut session, rejected) = match (first, second) {
+            (Ok(session), Err(error)) | (Err(error), Ok(session)) => (session, error),
+            (Ok(_), Ok(_)) => panic!("duplicate provisioning created two sessions"),
+            (Err(first), Err(second)) => {
+                panic!("both provisioning attempts failed: {first}; {second}")
+            }
+        };
+        assert!(matches!(rejected, WebRtcError::Signaling(_)));
+
+        session.shutdown().await.unwrap();
+        signaling.wait_closed(Duration::from_secs(2)).await.unwrap();
+        assert_eq!(session.snapshot().active_tasks, 0);
+        assert_eq!(signaling.active_tasks(), 0);
+    }
+
     #[tokio::test]
     async fn failed_answer_closes_provisioned_server_session() {
         let signaling = Arc::new(InvalidAnswerSignaling {

crates/libremetaverse/src/appearance_manager.rs

@@ -28,6 +28,7 @@ use libremetaverse_types::{
 use std::collections::{HashMap, HashSet, VecDeque};
 use std::fmt;
 use std::hash::{Hash, Hasher};
+use std::panic::{AssertUnwindSafe, catch_unwind};
 use std::sync::atomic::{AtomicBool, AtomicI32, AtomicU64, Ordering};
 use std::sync::{Arc, Mutex, RwLock};
 
@@ -84,7 +85,8 @@ impl<T: 'static> EventRegistry<T> {
     fn emit_with(&self, mut event: impl FnMut() -> T) {
         let handlers: Vec<_> = mutex(&self.handlers).values().cloned().collect();
         for handler in handlers {
-            handler(event());
+            let argument = event();
+            let _ = catch_unwind(AssertUnwindSafe(|| handler(argument)));
         }
     }
 }
@@ -2291,6 +2293,32 @@ fn inventory_item(value: &dyn InventoryObjectClass) -> Option<InventoryItem> {
 mod tests {
     use super::*;
 
+    #[test]
+    fn event_registry_isolates_panics_and_releases_subscriptions() {
+        let registry = EventRegistry::<usize>::default();
+        let released = Arc::new(());
+        let released_probe = Arc::downgrade(&released);
+        let panicking = registry.subscribe(Arc::new(|_| panic!("subscriber failure")));
+        let delivered = Arc::new(AtomicU64::new(0));
+        let delivered_handler = Arc::clone(&delivered);
+        let releasing = Arc::clone(&released);
+        let healthy = registry.subscribe(Arc::new(move |value| {
+            let _keep_alive = &releasing;
+            delivered_handler.fetch_add(value as u64, Ordering::AcqRel);
+        }));
+        drop(released);
+
+        registry.emit_with(|| 1);
+        assert_eq!(delivered.load(Ordering::Acquire), 1);
+        assert!(released_probe.upgrade().is_some());
+
+        drop(panicking);
+        drop(healthy);
+        assert!(released_probe.upgrade().is_none());
+        registry.emit_with(|| 1);
+        assert_eq!(delivered.load(Ordering::Acquire), 1);
+    }
+
     fn wearable(id: u64, wearable_type: WearableType, asset_type: AssetType) -> InventoryItem {
         let mut wearable = InventoryWearable::new(UUID::new_with_u_int64(id).unwrap()).unwrap();
         wearable.base.set_asset_type(asset_type);

crates/libremetaverse/src/asset_manager.rs

@@ -30,6 +30,7 @@ use libremetaverse_types::compat::{
 use libremetaverse_types::{AssetType, UUID, Utils};
 use serde_json::Value;
 use std::collections::{BTreeMap, HashMap};
+use std::panic::{AssertUnwindSafe, catch_unwind};
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
@@ -87,11 +88,43 @@ impl<T: Clone + 'static> EventSlot<T> {
     fn emit(&self, value: T) {
         let handlers: Vec<_> = mutex(&self.handlers).values().cloned().collect();
         for handler in handlers {
-            handler(value.clone());
+            let argument = value.clone();
+            let _ = catch_unwind(AssertUnwindSafe(|| handler(argument)));
         }
     }
 }
 
+#[cfg(test)]
+mod event_slot_tests {
+    use super::*;
+
+    #[test]
+    fn dispatch_isolates_panics_and_does_not_retain_dropped_handlers() {
+        let slot = Arc::new(EventSlot::<usize>::default());
+        let released = Arc::new(());
+        let released_probe = Arc::downgrade(&released);
+        let panicking = slot.subscribe(Arc::new(|_| panic!("subscriber failure")));
+        let delivered = Arc::new(AtomicU64::new(0));
+        let delivered_handler = Arc::clone(&delivered);
+        let releasing = Arc::clone(&released);
+        let healthy = slot.subscribe(Arc::new(move |value| {
+            let _keep_alive = &releasing;
+            delivered_handler.fetch_add(value as u64, Ordering::AcqRel);
+        }));
+        drop(released);
+
+        slot.emit(1);
+        assert_eq!(delivered.load(Ordering::Acquire), 1);
+        assert!(released_probe.upgrade().is_some());
+
+        drop(panicking);
+        drop(healthy);
+        assert!(released_probe.upgrade().is_none());
+        slot.emit(1);
+        assert_eq!(delivered.load(Ordering::Acquire), 1);
+    }
+}
+
 struct UploadState {
     asset_id: UUID,
     data: Vec<u8>,

crates/libremetaverse/src/client_core.rs

@@ -242,20 +242,25 @@ impl ClientRuntime {
         }
 
         self.cancellation.cancel();
-        let mut services = self
-            .services
-            .lock()
-            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        // Service shutdown is user-controlled code. Move registrations out
+        // before invoking it so re-entrant lifecycle calls cannot deadlock on
+        // the registry mutex.
+        let mut services = std::mem::take(
+            &mut *self
+                .services
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+        );
         services.sort_by_key(|entry| (entry.service.shutdown_phase(), entry.insertion_order));
         let mut first_error = None;
-        for entry in services.iter() {
+        for entry in &services {
             if let Err(error) = entry.service.shutdown()
                 && first_error.is_none()
             {
                 first_error = Some(error);
             }
         }
-        services.clear();
+        drop(services);
         clear_cached(&self.agent_throttle_sender);
         clear_cached(&self.inventory_manager);
         clear_cached(&self.inventory_ais_client);
@@ -1623,6 +1628,41 @@ mod tests {
         }
     }
 
+    struct ReentrantService {
+        client: Mutex<Option<GridClient>>,
+        calls: AtomicUsize,
+    }
+
+    impl ClientService for ReentrantService {
+        fn name(&self) -> &'static str {
+            "reentrant"
+        }
+
+        fn shutdown(&self) -> Result<(), ClientCoreError> {
+            self.calls.fetch_add(1, Ordering::AcqRel);
+            let result = self
+                .client
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .as_mut()
+                .expect("client installed")
+                .register_service(Arc::new(RecordingService {
+                    name: "late",
+                    phase: ShutdownPhase::Manager,
+                    order: Arc::new(Mutex::new(Vec::new())),
+                    calls: AtomicUsize::new(0),
+                }));
+            assert!(matches!(
+                result,
+                Err(ClientCoreError::InvalidLifecycle {
+                    state: ClientLifecycleState::ShuttingDown,
+                    ..
+                })
+            ));
+            Ok(())
+        }
+    }
+
     #[test]
     #[allow(clippy::float_cmp)] // Exact representable constants are the compatibility contract.
     fn defaults_match_reference_and_validate() {
@@ -1762,4 +1802,29 @@ mod tests {
         assert_eq!(network.calls.load(Ordering::Relaxed), 1);
         assert_eq!(manager.calls.load(Ordering::Relaxed), 1);
     }
+
+    #[test]
+    fn shutdown_does_not_hold_service_registry_lock_across_callbacks() {
+        let service = Arc::new(ReentrantService {
+            client: Mutex::new(None),
+            calls: AtomicUsize::new(0),
+        });
+        let client = GridClient::builder()
+            .with_service(service.clone())
+            .build()
+            .unwrap();
+        *service
+            .client
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = Some(client.clone());
+
+        client.shutdown().unwrap();
+        assert_eq!(client.lifecycle_state(), ClientLifecycleState::Disposed);
+        assert_eq!(service.calls.load(Ordering::Acquire), 1);
+        service
+            .client
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .take();
+    }
 }

crates/libremetaverse/src/download_manager.rs

@@ -511,11 +511,21 @@ fn download_dispatcher(
                 biased;
                 () = shutdown.cancelled() => break,
                 Some(job) = receiver.recv() => {
-                    let manager = manager.clone();
-                    let gate = Arc::clone(&gate);
-                    jobs.spawn(async move {
-                        run_download_job(job, manager, gate).await;
-                    });
+                    let cancellation = job.active.cancellation.token();
+                    match gate.acquire(cancellation).await {
+                        Ok(permit) => {
+                            let manager = manager.clone();
+                            jobs.spawn(async move {
+                                run_download_job(job, manager, permit).await;
+                            });
+                        }
+                        Err(error) => {
+                            if let Some(manager) = manager.upgrade() {
+                                manager.remove_active(&job.key, &job.active);
+                            }
+                            job.active.complete(Err(error));
+                        }
+                    }
                 }
                 Some(_) = jobs.join_next(), if !jobs.is_empty() => {}
                 else => break,
@@ -534,14 +544,10 @@ fn download_dispatcher(
 async fn run_download_job(
     job: DownloadJob,
     manager: Weak<DownloadManagerInner>,
-    gate: Arc<DynamicGate>,
+    _permit: GatePermit,
 ) {
     let cancellation = job.active.cancellation.token();
-    let permit = gate.acquire(cancellation.clone()).await;
-    let result = match permit {
-        Ok(_permit) => perform_download(&job, cancellation).await,
-        Err(error) => Err(error),
-    };
+    let result = perform_download(&job, cancellation).await;
     if let Some(manager) = manager.upgrade() {
         manager.remove_active(&job.key, &job.active);
     }

crates/libremetaverse/src/inventory_manager.rs

@@ -774,6 +774,35 @@ impl fmt::Debug for InventoryManager {
 }
 
 impl InventoryManager {
+    /// Whether the manager's callback-expiry worker is still owned and may run.
+    #[must_use]
+    pub fn cleanup_worker_running(&self) -> bool {
+        mutex(&self.inner.cleanup_worker).is_some()
+    }
+
+    /// Total asynchronous inventory operations still awaiting a reply.
+    #[must_use]
+    pub fn pending_operation_count(&self) -> usize {
+        mutex(&self.inner.pending_created).len()
+            + mutex(&self.inner.pending_copied).len()
+            + mutex(&self.inner.pending_fetches)
+                .values()
+                .map(Vec::len)
+                .sum::<usize>()
+            + mutex(&self.inner.pending_task_replies)
+                .values()
+                .map(Vec::len)
+                .sum::<usize>()
+            + mutex(&self.inner.pending_offers).len()
+            + mutex(&self.inner.task_xfers).len()
+    }
+
+    /// Whether deterministic manager teardown has completed.
+    #[must_use]
+    pub fn is_disposed(&self) -> bool {
+        self.inner.disposed.load(Ordering::Acquire)
+    }
+
     pub(crate) fn native_from_inner(inner: Arc<InventoryManagerInner>) -> Self {
         Self { inner }
     }

crates/libremetaverse/tests/caps_http.rs

@@ -723,6 +723,67 @@ async fn downloads_honor_parallel_limit_progress_and_external_completion_source(
     downloads.dispose().expect("dispose");
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn download_queue_applies_backpressure_and_cancellation_drains_every_job() {
+    let active = Arc::new(AtomicUsize::new(0));
+    let handler_active = Arc::clone(&active);
+    let handler = HttpMessageHandler::new(move |_request, cancellation| {
+        let active = Arc::clone(&handler_active);
+        async move {
+            active.fetch_add(1, Ordering::AcqRel);
+            cancellation.cancelled().await;
+            active.fetch_sub(1, Ordering::AcqRel);
+            response(200, Vec::new())
+        }
+    });
+    let mut client = GridClient::new().expect("client");
+    client.set_http_caps_client(HttpCapsClient::new(handler).expect("HTTP client"));
+    let mut downloads = DownloadManager::new(client).expect("downloads");
+    downloads.set_parallel_downloads(1);
+    downloads
+        .queue_download_with_download_request(
+            DownloadRequest::new(
+                Uri("http://example.test/backpressure/initial".into()),
+                None,
+                None,
+            )
+            .expect("request"),
+        )
+        .expect("initial request");
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while active.load(Ordering::Acquire) == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("initial request started");
+
+    let mut rejected = 0;
+    for index in 0..300 {
+        let result = downloads.queue_download_with_download_request(
+            DownloadRequest::new(
+                Uri(format!("http://example.test/backpressure/{index}")),
+                None,
+                None,
+            )
+            .expect("request"),
+        );
+        match result {
+            Ok(()) => {}
+            Err(Error::InvalidOperation) => rejected += 1,
+            Err(error) => panic!("unexpected queue error: {error:?}"),
+        }
+    }
+    assert!(rejected > 0, "the bounded queue must reject excess work");
+    assert!(downloads.active_download_count() <= 258);
+
+    downloads.dispose().expect("dispose");
+    assert_eq!(active.load(Ordering::Acquire), 0);
+    assert_eq!(downloads.active_download_count(), 0);
+    assert!(!downloads.dispatcher_running());
+    assert!(downloads.is_disposed());
+}
+
 async fn read_http_request(stream: &mut tokio::net::TcpStream) -> Vec<u8> {
     let mut request = Vec::new();
     let mut scratch = [0_u8; 1024];