Implement capability HTTP and downloads (#54)

crates/libremetaverse/src/download_manager.rs

@@ -0,0 +1,609 @@
+//! Bounded, deduplicating HTTP download dispatcher.
+
+#![allow(clippy::missing_errors_doc)] // Public result shapes are fixed by the compatibility map.
+#![allow(clippy::needless_pass_by_value)] // Mapped APIs preserve owned CLR argument shapes.
+
+use crate::{Error, GridClient, HttpCapsClientProgressReport};
+use libremetaverse_types::compat::{
+    CancellationToken, CancellationTokenSource, HttpResponse, IProgress, Subscription,
+    TaskCompletionSource, Uri,
+};
+use std::collections::HashMap;
+use std::fmt;
+use std::panic::{AssertUnwindSafe, catch_unwind};
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex, Weak};
+use std::thread::{self, JoinHandle};
+use std::time::Duration;
+use tokio::sync::{Notify, mpsc, oneshot};
+use tokio::task::JoinSet;
+
+const DEFAULT_PARALLEL_DOWNLOADS: usize = 8;
+const MAX_PARALLEL_DOWNLOADS: usize = 32;
+const DOWNLOAD_QUEUE_CAPACITY: usize = 256;
+
+type DownloadValue = (HttpResponse, Vec<u8>);
+type DownloadCompletion = TaskCompletionSource<DownloadValue>;
+type DownloadResult = Result<DownloadValue, Error>;
+
+fn mutex<T>(value: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
+    value
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+/// Mapped request state with the same defaults as the C# constructor.
+pub struct DownloadRequest {
+    pub address: Uri,
+    pub attempt: i32,
+    pub cancellation_token: CancellationToken,
+    pub completion_tcs: Option<DownloadCompletion>,
+    pub content_type: Option<String>,
+    pub download_progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
+    pub retries: i32,
+}
+
+impl DownloadRequest {
+    pub fn new(
+        address: Uri,
+        content_type: Option<String>,
+        download_progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
+    ) -> Result<Self, Error> {
+        validate_http_uri(&address)?;
+        Ok(Self {
+            address,
+            attempt: 0,
+            cancellation_token: CancellationToken::default(),
+            completion_tcs: None,
+            content_type,
+            download_progress_callback,
+            retries: 5,
+        })
+    }
+}
+
+impl fmt::Debug for DownloadRequest {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("DownloadRequest")
+            .field("address", &"<redacted capability URI>")
+            .field("attempt", &self.attempt)
+            .field(
+                "cancellation_requested",
+                &self.cancellation_token.is_cancellation_requested(),
+            )
+            .field("has_completion", &self.completion_tcs.is_some())
+            .field("has_content_type", &self.content_type.is_some())
+            .field("has_progress", &self.download_progress_callback.is_some())
+            .field("retries", &self.retries)
+            .finish()
+    }
+}
+
+struct ActiveDownload {
+    cancellation: CancellationTokenSource,
+    cancellation_guards: Mutex<Vec<Subscription>>,
+    progress: Mutex<Vec<Arc<dyn IProgress<HttpCapsClientProgressReport>>>>,
+    completion_sources: Mutex<Vec<DownloadCompletion>>,
+    waiters: Mutex<Vec<oneshot::Sender<DownloadResult>>>,
+    completed: AtomicBool,
+}
+
+impl ActiveDownload {
+    fn new() -> Self {
+        Self {
+            cancellation: CancellationTokenSource::new(),
+            cancellation_guards: Mutex::new(Vec::new()),
+            progress: Mutex::new(Vec::new()),
+            completion_sources: Mutex::new(Vec::new()),
+            waiters: Mutex::new(Vec::new()),
+            completed: AtomicBool::new(false),
+        }
+    }
+
+    fn attach_cancellation(&self, token: &CancellationToken) {
+        let cancellation = self.cancellation.clone();
+        let guard = token.register_callback(Arc::new(move || cancellation.cancel()));
+        mutex(&self.cancellation_guards).push(guard);
+    }
+
+    fn attach_progress(&self, progress: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>) {
+        if let Some(progress) = progress {
+            mutex(&self.progress).push(Arc::from(progress));
+        }
+    }
+
+    fn add_waiter(&self) -> oneshot::Receiver<DownloadResult> {
+        let (sender, receiver) = oneshot::channel();
+        mutex(&self.waiters).push(sender);
+        receiver
+    }
+
+    fn attach_completion_source(&self, completion_source: Option<DownloadCompletion>) {
+        if let Some(completion_source) = completion_source {
+            mutex(&self.completion_sources).push(completion_source);
+        }
+    }
+
+    fn complete(&self, result: DownloadResult) {
+        if self.completed.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        for waiter in std::mem::take(&mut *mutex(&self.waiters)) {
+            let _ = waiter.send(result.clone());
+        }
+        for completion_source in std::mem::take(&mut *mutex(&self.completion_sources)) {
+            match result.clone() {
+                Ok(value) => {
+                    let _ = completion_source.try_set_result(value);
+                }
+                Err(Error::Cancelled) => {
+                    let _ = completion_source.try_set_cancelled();
+                }
+                Err(error) => {
+                    let _ = completion_source.try_set_error(error);
+                }
+            }
+        }
+        mutex(&self.progress).clear();
+        mutex(&self.cancellation_guards).clear();
+    }
+
+    fn report(&self, report: HttpCapsClientProgressReport) {
+        let handlers = mutex(&self.progress).clone();
+        for handler in handlers {
+            let _ = catch_unwind(AssertUnwindSafe(|| handler.report(report)));
+        }
+    }
+}
+
+struct ProgressFanout(Weak<ActiveDownload>);
+
+impl IProgress<HttpCapsClientProgressReport> for ProgressFanout {
+    fn report(&self, report: HttpCapsClientProgressReport) {
+        if let Some(active) = self.0.upgrade() {
+            active.report(report);
+        }
+    }
+}
+
+struct DownloadJob {
+    key: String,
+    address: Uri,
+    attempt: i32,
+    retries: i32,
+    active: Arc<ActiveDownload>,
+    client: GridClient,
+}
+
+struct GateState {
+    active: usize,
+}
+
+struct DynamicGate {
+    limit: AtomicUsize,
+    state: Mutex<GateState>,
+    notify: Notify,
+}
+
+impl DynamicGate {
+    fn new(limit: usize) -> Self {
+        Self {
+            limit: AtomicUsize::new(limit),
+            state: Mutex::new(GateState { active: 0 }),
+            notify: Notify::new(),
+        }
+    }
+
+    fn set_limit(&self, value: usize) {
+        self.limit.store(value, Ordering::Release);
+        self.notify.notify_waiters();
+    }
+
+    async fn acquire(
+        self: &Arc<Self>,
+        cancellation: CancellationToken,
+    ) -> Result<GatePermit, Error> {
+        loop {
+            cancellation.throw_if_cancellation_requested()?;
+            let notified = self.notify.notified();
+            {
+                let mut state = mutex(&self.state);
+                if state.active < self.limit.load(Ordering::Acquire) {
+                    state.active += 1;
+                    return Ok(GatePermit(Arc::clone(self)));
+                }
+            }
+            tokio::select! {
+                () = notified => {}
+                () = cancellation.cancelled() => return Err(Error::Cancelled),
+            }
+        }
+    }
+}
+
+struct GatePermit(Arc<DynamicGate>);
+
+impl Drop for GatePermit {
+    fn drop(&mut self) {
+        let mut state = mutex(&self.0.state);
+        state.active = state.active.saturating_sub(1);
+        drop(state);
+        self.0.notify.notify_one();
+    }
+}
+
+struct DownloadManagerInner {
+    client: GridClient,
+    active: Mutex<HashMap<String, Arc<ActiveDownload>>>,
+    sender: Mutex<Option<mpsc::Sender<DownloadJob>>>,
+    dispatcher: Mutex<Option<JoinHandle<()>>>,
+    shutdown: CancellationTokenSource,
+    gate: Arc<DynamicGate>,
+    parallel_downloads: AtomicUsize,
+    disposed: AtomicBool,
+}
+
+impl DownloadManagerInner {
+    fn remove_active(&self, key: &str, active: &Arc<ActiveDownload>) {
+        let mut downloads = mutex(&self.active);
+        if downloads
+            .get(key)
+            .is_some_and(|candidate| Arc::ptr_eq(candidate, active))
+        {
+            downloads.remove(key);
+        }
+    }
+
+    fn shutdown(&self) {
+        if self.disposed.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        self.shutdown.cancel();
+        for active in mutex(&self.active).values() {
+            active.cancellation.cancel();
+        }
+        mutex(&self.sender).take();
+        if let Some(dispatcher) = mutex(&self.dispatcher).take()
+            && dispatcher.thread().id() != thread::current().id()
+        {
+            let _ = dispatcher.join();
+        }
+        let remaining = std::mem::take(&mut *mutex(&self.active));
+        for active in remaining.into_values() {
+            active.complete(Err(Error::Cancelled));
+        }
+    }
+}
+
+impl Drop for DownloadManagerInner {
+    fn drop(&mut self) {
+        self.shutdown();
+    }
+}
+
+/// Native, bounded download manager with URI-level request deduplication.
+#[derive(Clone)]
+pub struct DownloadManager(Arc<DownloadManagerInner>);
+
+impl DownloadManager {
+    pub fn new(client: GridClient) -> Result<Self, Error> {
+        let (sender, receiver) = mpsc::channel(DOWNLOAD_QUEUE_CAPACITY);
+        let gate = Arc::new(DynamicGate::new(DEFAULT_PARALLEL_DOWNLOADS));
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .map_err(|_| Error::InvalidOperation)?;
+        let inner = Arc::new(DownloadManagerInner {
+            client,
+            active: Mutex::new(HashMap::new()),
+            sender: Mutex::new(Some(sender)),
+            dispatcher: Mutex::new(None),
+            shutdown: CancellationTokenSource::new(),
+            gate: Arc::clone(&gate),
+            parallel_downloads: AtomicUsize::new(DEFAULT_PARALLEL_DOWNLOADS),
+            disposed: AtomicBool::new(false),
+        });
+        let weak = Arc::downgrade(&inner);
+        let shutdown = inner.shutdown.token();
+        let dispatcher = thread::Builder::new()
+            .name("libremetaverse-download-dispatcher".to_owned())
+            .spawn(move || download_dispatcher(runtime, receiver, weak, gate, shutdown))
+            .map_err(|_| Error::InvalidOperation)?;
+        *mutex(&inner.dispatcher) = Some(dispatcher);
+        Ok(Self(inner))
+    }
+
+    pub fn dispose(&self) -> Result<(), Error> {
+        self.0.shutdown();
+        Ok(())
+    }
+
+    #[must_use]
+    pub fn parallel_downloads(&self) -> i32 {
+        i32::try_from(self.0.parallel_downloads.load(Ordering::Acquire)).unwrap_or(i32::MAX)
+    }
+
+    pub fn set_parallel_downloads(&mut self, value: i32) {
+        let value = usize::try_from(value)
+            .unwrap_or(1)
+            .clamp(1, MAX_PARALLEL_DOWNLOADS);
+        self.0.parallel_downloads.store(value, Ordering::Release);
+        self.0.gate.set_limit(value);
+    }
+
+    pub fn queue_download_with_download_request(
+        &self,
+        request: DownloadRequest,
+    ) -> Result<(), Error> {
+        self.enqueue(request, false).map(|_| ())
+    }
+
+    pub fn queue_download_with_download_request_cancellation_token(
+        &self,
+        mut request: DownloadRequest,
+        cancellation_token: CancellationToken,
+    ) -> Result<(), Error> {
+        request.cancellation_token = cancellation_token;
+        self.queue_download_with_download_request(request)
+    }
+
+    pub async fn queue_download_with_download_request_ba727387(
+        &self,
+        request: DownloadRequest,
+    ) -> DownloadResult {
+        let cancellation = request.cancellation_token.clone();
+        let (active, receiver) = self.enqueue(request, true)?;
+        await_download(
+            active,
+            receiver.ok_or(Error::InvalidOperation)?,
+            cancellation,
+        )
+        .await
+    }
+
+    pub async fn queue_download_with_uri_string_i_progress_cancellation_token_int32(
+        &self,
+        address: Uri,
+        content_type: Option<String>,
+        progress_callback: Option<Box<dyn IProgress<HttpCapsClientProgressReport>>>,
+        cancellation_token: Option<CancellationToken>,
+        retries: Option<i32>,
+    ) -> DownloadResult {
+        let cancellation = cancellation_token.unwrap_or_default();
+        let mut request = DownloadRequest::new(address, content_type, progress_callback)?;
+        request.cancellation_token = cancellation.clone();
+        request.retries = retries.unwrap_or(5).max(0);
+        let (active, receiver) = self.enqueue(request, true)?;
+        await_download(
+            active,
+            receiver.ok_or(Error::InvalidOperation)?,
+            cancellation,
+        )
+        .await
+    }
+
+    pub async fn download_with_uri_string_i_progress_cancellation_token(
+        &self,
+        address: Uri,
+        content_type: String,
+        progress: Box<dyn IProgress<HttpCapsClientProgressReport>>,
+        cancellation_token: CancellationToken,
+    ) -> DownloadResult {
+        self.queue_download_with_uri_string_i_progress_cancellation_token_int32(
+            address,
+            Some(content_type),
+            Some(progress),
+            Some(cancellation_token),
+            Some(5),
+        )
+        .await
+    }
+
+    pub async fn download_with_uri_i_progress_cancellation_token(
+        &self,
+        address: Uri,
+        progress: Box<dyn IProgress<HttpCapsClientProgressReport>>,
+        cancellation_token: CancellationToken,
+    ) -> DownloadResult {
+        self.download_with_uri_string_i_progress_cancellation_token(
+            address,
+            String::new(),
+            progress,
+            cancellation_token,
+        )
+        .await
+    }
+
+    fn enqueue(
+        &self,
+        mut request: DownloadRequest,
+        with_waiter: bool,
+    ) -> Result<
+        (
+            Arc<ActiveDownload>,
+            Option<oneshot::Receiver<DownloadResult>>,
+        ),
+        Error,
+    > {
+        if self.0.disposed.load(Ordering::Acquire) {
+            return Err(Error::InvalidOperation);
+        }
+        let key = validate_http_uri(&request.address)?;
+        let selected = {
+            let mut active = mutex(&self.0.active);
+            if let Some(existing) = active.get(&key) {
+                Some((Arc::clone(existing), false))
+            } else if request.cancellation_token.is_cancellation_requested() {
+                None
+            } else {
+                let download = Arc::new(ActiveDownload::new());
+                active.insert(key.clone(), Arc::clone(&download));
+                Some((download, true))
+            }
+        };
+        let Some((active, is_new)) = selected else {
+            if let Some(completion_source) = request.completion_tcs.take() {
+                let _ = completion_source.try_set_cancelled();
+            }
+            return Err(Error::Cancelled);
+        };
+        active.attach_cancellation(&request.cancellation_token);
+        active.attach_progress(request.download_progress_callback.take());
+        active.attach_completion_source(request.completion_tcs.take());
+        let receiver = with_waiter.then(|| active.add_waiter());
+        if !is_new {
+            return Ok((active, receiver));
+        }
+
+        let job = DownloadJob {
+            key: key.clone(),
+            address: request.address,
+            attempt: request.attempt.max(0),
+            retries: request.retries.max(0),
+            active: Arc::clone(&active),
+            client: self.0.client.clone(),
+        };
+        let sender = mutex(&self.0.sender)
+            .clone()
+            .ok_or(Error::InvalidOperation)?;
+        if sender.try_send(job).is_err() {
+            self.0.remove_active(&key, &active);
+            active.complete(Err(Error::InvalidOperation));
+            return Err(Error::InvalidOperation);
+        }
+        Ok((active, receiver))
+    }
+}
+
+impl fmt::Debug for DownloadManager {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("DownloadManager")
+            .field("active_downloads", &mutex(&self.0.active).len())
+            .field("parallel_downloads", &self.parallel_downloads())
+            .field("queue_capacity", &DOWNLOAD_QUEUE_CAPACITY)
+            .field("disposed", &self.0.disposed.load(Ordering::Acquire))
+            .finish()
+    }
+}
+
+async fn await_download(
+    active: Arc<ActiveDownload>,
+    receiver: oneshot::Receiver<DownloadResult>,
+    cancellation: CancellationToken,
+) -> DownloadResult {
+    tokio::select! {
+        result = receiver => result.map_err(|_| Error::Cancelled)?,
+        () = cancellation.cancelled() => {
+            active.cancellation.cancel();
+            Err(Error::Cancelled)
+        }
+    }
+}
+
+fn download_dispatcher(
+    runtime: tokio::runtime::Runtime,
+    mut receiver: mpsc::Receiver<DownloadJob>,
+    manager: Weak<DownloadManagerInner>,
+    gate: Arc<DynamicGate>,
+    shutdown: CancellationToken,
+) {
+    runtime.block_on(async move {
+        let mut jobs = JoinSet::new();
+        loop {
+            tokio::select! {
+                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;
+                    });
+                }
+                Some(_) = jobs.join_next(), if !jobs.is_empty() => {}
+                else => break,
+            }
+        }
+        while let Ok(job) = receiver.try_recv() {
+            job.active.complete(Err(Error::Cancelled));
+            if let Some(manager) = manager.upgrade() {
+                manager.remove_active(&job.key, &job.active);
+            }
+        }
+        while jobs.join_next().await.is_some() {}
+    });
+}
+
+async fn run_download_job(
+    job: DownloadJob,
+    manager: Weak<DownloadManagerInner>,
+    gate: Arc<DynamicGate>,
+) {
+    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),
+    };
+    if let Some(manager) = manager.upgrade() {
+        manager.remove_active(&job.key, &job.active);
+    }
+    job.active.complete(result);
+}
+
+async fn perform_download(job: &DownloadJob, cancellation: CancellationToken) -> DownloadResult {
+    let mut attempt = job.attempt;
+    loop {
+        cancellation.throw_if_cancellation_requested()?;
+        let progress: Box<dyn IProgress<HttpCapsClientProgressReport>> =
+            Box::new(ProgressFanout(Arc::downgrade(&job.active)));
+        let result = job
+            .client
+            .native_http_caps_client()
+            .get(job.address.clone(), cancellation.clone(), Some(progress))
+            .await;
+        match result {
+            Ok((response, data)) if response.is_success_status_code() => {
+                return Ok((response, data));
+            }
+            Ok((response, _)) => {
+                if is_permanent_status(response.status_code) || attempt >= job.retries {
+                    return Err(Error::HttpRequest);
+                }
+            }
+            Err(Error::Cancelled) => return Err(Error::Cancelled),
+            Err(error) if attempt >= job.retries => return Err(error),
+            Err(_) => {}
+        }
+        attempt = attempt.saturating_add(1);
+        let base_delay = u64::try_from(200_i32.saturating_mul(attempt).min(2_000)).unwrap_or(2_000);
+        let delay =
+            Duration::from_millis(base_delay.saturating_add(retry_jitter(&job.key, attempt)));
+        tokio::select! {
+            () = tokio::time::sleep(delay) => {}
+            () = cancellation.cancelled() => return Err(Error::Cancelled),
+        }
+    }
+}
+
+fn is_permanent_status(status: u16) -> bool {
+    matches!(status, 401 | 403 | 404 | 410)
+}
+
+fn retry_jitter(key: &str, attempt: i32) -> u64 {
+    let hash = key.bytes().fold(
+        0xcbf2_9ce4_8422_2325_u64 ^ u64::try_from(attempt).unwrap_or_default(),
+        |hash, byte| hash.wrapping_mul(0x0000_0100_0000_01b3) ^ u64::from(byte),
+    );
+    hash % 200
+}
+
+fn validate_http_uri(uri: &Uri) -> Result<String, Error> {
+    let parsed = reqwest::Url::parse(&uri.0).map_err(|_| Error::Argument)?;
+    if !matches!(parsed.scheme(), "http" | "https") || parsed.host().is_none() {
+        return Err(Error::Argument);
+    }
+    Ok(parsed.to_string())
+}