Implement native client core lifecycle (#51)

crates/libremetaverse-types/src/compat.rs

@@ -12,9 +12,10 @@ use std::hash::Hash;
 use std::marker::PhantomData;
 use std::pin::Pin;
 use std::sync::{
-    Arc,
-    atomic::{AtomicBool, Ordering},
+    Arc, Mutex, Weak,
+    atomic::{AtomicBool, AtomicU64, Ordering},
 };
+use std::task::{Context, Poll, Waker};
 
 pub trait Collection<T> {}
 
@@ -289,13 +290,179 @@ pub struct XmlTextReader(pub String);
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct XmlTextWriter(pub String);
 
-#[derive(Clone, Debug, Default)]
-pub struct CancellationToken(Arc<AtomicBool>);
+struct CancellationState {
+    cancelled: AtomicBool,
+    next_id: AtomicU64,
+    waiters: Mutex<BTreeMap<u64, Waker>>,
+    callbacks: Mutex<BTreeMap<u64, Arc<dyn Fn() + Send + Sync>>>,
+}
+
+impl Default for CancellationState {
+    fn default() -> Self {
+        Self {
+            cancelled: AtomicBool::new(false),
+            next_id: AtomicU64::new(1),
+            waiters: Mutex::new(BTreeMap::new()),
+            callbacks: Mutex::new(BTreeMap::new()),
+        }
+    }
+}
+
+/// A cheap, cloneable cancellation signal.
+///
+/// Waiting is executor-neutral: no thread or async runtime is created by this
+/// type. Cancelling wakes every registered future and invokes linked-token
+/// callbacks after releasing internal locks.
+#[derive(Clone, Default)]
+pub struct CancellationToken(Arc<CancellationState>);
+
+impl fmt::Debug for CancellationToken {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("CancellationToken")
+            .field(
+                "is_cancellation_requested",
+                &self.is_cancellation_requested(),
+            )
+            .finish_non_exhaustive()
+    }
+}
 
 impl CancellationToken {
     #[must_use]
     pub fn is_cancellation_requested(&self) -> bool {
-        self.0.load(Ordering::Acquire)
+        self.0.cancelled.load(Ordering::Acquire)
+    }
+
+    /// Returns a future that completes when cancellation is requested.
+    #[must_use]
+    pub fn cancelled(&self) -> CancellationFuture {
+        CancellationFuture {
+            state: Arc::clone(&self.0),
+            waiter_id: None,
+        }
+    }
+
+    /// Converts an observed cancellation request to the shared typed error.
+    ///
+    /// # Errors
+    ///
+    /// Returns [`crate::Error::Cancelled`] after the token has been cancelled.
+    pub fn throw_if_cancellation_requested(&self) -> Result<(), crate::Error> {
+        if self.is_cancellation_requested() {
+            Err(crate::Error::Cancelled)
+        } else {
+            Ok(())
+        }
+    }
+
+    fn register(&self, callback: Arc<dyn Fn() + Send + Sync>) -> CancellationRegistration {
+        if self.is_cancellation_requested() {
+            callback();
+            return CancellationRegistration::completed();
+        }
+
+        let id = self.0.next_id.fetch_add(1, Ordering::Relaxed);
+        self.0
+            .callbacks
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(id, callback);
+        if self.is_cancellation_requested() {
+            let callback = self
+                .0
+                .callbacks
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .remove(&id);
+            if let Some(callback) = callback {
+                callback();
+            }
+            CancellationRegistration::completed()
+        } else {
+            CancellationRegistration {
+                state: Arc::downgrade(&self.0),
+                id: Some(id),
+            }
+        }
+    }
+}
+
+/// Future returned by [`CancellationToken::cancelled`].
+pub struct CancellationFuture {
+    state: Arc<CancellationState>,
+    waiter_id: Option<u64>,
+}
+
+impl Future for CancellationFuture {
+    type Output = ();
+
+    fn poll(mut self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Self::Output> {
+        if self.state.cancelled.load(Ordering::Acquire) {
+            return Poll::Ready(());
+        }
+
+        let id = if let Some(id) = self.waiter_id {
+            id
+        } else {
+            let id = self.state.next_id.fetch_add(1, Ordering::Relaxed);
+            self.waiter_id = Some(id);
+            id
+        };
+        self.state
+            .waiters
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .insert(id, context.waker().clone());
+        if self.state.cancelled.load(Ordering::Acquire) {
+            self.state
+                .waiters
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .remove(&id);
+            self.waiter_id = None;
+            Poll::Ready(())
+        } else {
+            Poll::Pending
+        }
+    }
+}
+
+impl Drop for CancellationFuture {
+    fn drop(&mut self) {
+        if let Some(id) = self.waiter_id {
+            self.state
+                .waiters
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .remove(&id);
+        }
+    }
+}
+
+struct CancellationRegistration {
+    state: Weak<CancellationState>,
+    id: Option<u64>,
+}
+
+impl CancellationRegistration {
+    fn completed() -> Self {
+        Self {
+            state: Weak::new(),
+            id: None,
+        }
+    }
+}
+
+impl Drop for CancellationRegistration {
+    fn drop(&mut self) {
+        if let (Some(state), Some(id)) = (self.state.upgrade(), self.id) {
+            state
+                .callbacks
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner)
+                .remove(&id);
+        }
     }
 }
 
@@ -310,8 +477,35 @@ pub struct DictionaryEntry(pub Object, pub Object);
 
 pub struct RateLimitLease;
 
-#[derive(Clone, Debug, Default)]
-pub struct CancellationTokenSource(CancellationToken);
+struct CancellationSourceState {
+    token: CancellationToken,
+    linked_registrations: Mutex<Vec<CancellationRegistration>>,
+}
+
+/// Owns and triggers a [`CancellationToken`].
+#[derive(Clone)]
+pub struct CancellationTokenSource(Arc<CancellationSourceState>);
+
+impl Default for CancellationTokenSource {
+    fn default() -> Self {
+        Self(Arc::new(CancellationSourceState {
+            token: CancellationToken::default(),
+            linked_registrations: Mutex::new(Vec::new()),
+        }))
+    }
+}
+
+impl fmt::Debug for CancellationTokenSource {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("CancellationTokenSource")
+            .field(
+                "is_cancellation_requested",
+                &self.0.token.is_cancellation_requested(),
+            )
+            .finish_non_exhaustive()
+    }
+}
 
 impl CancellationTokenSource {
     #[must_use]
@@ -319,13 +513,62 @@ impl CancellationTokenSource {
         Self::default()
     }
 
+    /// Creates a source cancelled when any parent token is cancelled.
+    #[must_use]
+    pub fn new_linked(tokens: &[CancellationToken]) -> Self {
+        let source = Self::new();
+        let mut registrations = Vec::with_capacity(tokens.len());
+        for token in tokens {
+            let linked = Arc::downgrade(&source.0);
+            registrations.push(token.register(Arc::new(move || {
+                if let Some(linked) = linked.upgrade() {
+                    Self::cancel_state(&linked);
+                }
+            })));
+        }
+        *source
+            .0
+            .linked_registrations
+            .lock()
+            .unwrap_or_else(std::sync::PoisonError::into_inner) = registrations;
+        source
+    }
+
     #[must_use]
     pub fn token(&self) -> CancellationToken {
-        self.0.clone()
+        self.0.token.clone()
     }
 
     pub fn cancel(&self) {
-        self.0.0.store(true, Ordering::Release);
+        Self::cancel_state(&self.0);
+    }
+
+    fn cancel_state(state: &CancellationSourceState) {
+        if state.token.0.cancelled.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        let waiters = std::mem::take(
+            &mut *state
+                .token
+                .0
+                .waiters
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+        );
+        let callbacks = std::mem::take(
+            &mut *state
+                .token
+                .0
+                .callbacks
+                .lock()
+                .unwrap_or_else(std::sync::PoisonError::into_inner),
+        );
+        for (_, waker) in waiters {
+            waker.wake();
+        }
+        for (_, callback) in callbacks {
+            callback();
+        }
     }
 }
 
@@ -347,7 +590,43 @@ pub struct Task<T>(pub PhantomData<fn() -> T>);
 
 pub struct TaskCompletionSource<T>(pub PhantomData<fn(T)>);
 
-pub struct TimeProvider;
+/// Cloneable clock boundary used by runtime-neutral client code.
+#[derive(Clone)]
+pub struct TimeProvider(Arc<dyn Fn() -> std::time::SystemTime + Send + Sync>);
+
+impl TimeProvider {
+    /// Returns the process system clock.
+    #[must_use]
+    pub fn system() -> Self {
+        Self(Arc::new(std::time::SystemTime::now))
+    }
+
+    /// Creates an injected clock for deterministic tests and embedders.
+    #[must_use]
+    pub fn from_fn<F>(clock: F) -> Self
+    where
+        F: Fn() -> std::time::SystemTime + Send + Sync + 'static,
+    {
+        Self(Arc::new(clock))
+    }
+
+    #[must_use]
+    pub fn get_utc_now(&self) -> std::time::SystemTime {
+        (self.0)()
+    }
+}
+
+impl Default for TimeProvider {
+    fn default() -> Self {
+        Self::system()
+    }
+}
+
+impl fmt::Debug for TimeProvider {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("TimeProvider(<injected clock>)")
+    }
+}
 
 pub struct MediaTypeHeaderValue(pub String);
 
@@ -459,11 +738,44 @@ pub struct SocketException;
 
 #[cfg(test)]
 mod tests {
-    use super::{CancellationToken, HttpMessageHandler, HttpRequest, HttpResponse, Object, Uri};
+    use super::{
+        CancellationToken, CancellationTokenSource, HttpMessageHandler, HttpRequest, HttpResponse,
+        Object, TimeProvider, Uri,
+    };
     use std::collections::BTreeMap;
     use std::future::Future;
     use std::sync::Arc;
     use std::task::{Context, Poll, Waker};
+    use std::time::{Duration, SystemTime};
+
+    #[test]
+    fn cancellation_wakes_waiters_and_propagates_from_linked_tokens() {
+        let parent = CancellationTokenSource::new();
+        let linked = CancellationTokenSource::new_linked(&[parent.token()]);
+        let mut future = Box::pin(linked.token().cancelled());
+        let mut context = Context::from_waker(Waker::noop());
+        assert!(matches!(future.as_mut().poll(&mut context), Poll::Pending));
+
+        parent.cancel();
+        assert!(linked.token().is_cancellation_requested());
+        assert!(matches!(
+            future.as_mut().poll(&mut context),
+            Poll::Ready(())
+        ));
+        assert_eq!(
+            linked.token().throw_if_cancellation_requested(),
+            Err(crate::Error::Cancelled)
+        );
+        parent.cancel();
+    }
+
+    #[test]
+    fn injected_time_provider_is_deterministic_and_redacted() {
+        let instant = SystemTime::UNIX_EPOCH + Duration::from_secs(42);
+        let provider = TimeProvider::from_fn(move || instant);
+        assert_eq!(provider.get_utc_now(), instant);
+        assert!(!format!("{provider:?}").contains("42"));
+    }
 
     #[test]
     fn http_handler_preserves_request_and_response_data() {