Download manager

Cargo.toml

@@ -13,6 +13,7 @@ readme = "README.md"
 
 [workspace]
 members = [
+    "crates/download-manager",
     "crates/next-cli",
     "crates/next-core",
     "crates/next-server",
@@ -28,7 +29,11 @@ thiserror = "2.0.15"
 tracing = "0.1.41"
 tonic = "0.14"
 tokio = { version = "1.47", features = ["full"] }
+tokio-stream = { version = "0.1.17", features = ["full"] }
+tokio-util = { version = "0.7", features = ["full"] }
 prost = "0.14"
 reqwest = { version = "0.12", features = ["stream"] }
 anyhow = "1.0.99"
 futures-core = "0.3"
+futures-util = "0.3"
+derive_builder = "0.20"

crates/download-manager/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "download-manager"
+version = "0.1.0"
+edition.workspace = true
+
+[lib]
+path = "src/download_manager.rs"
+
+[dependencies]
+reqwest = { workspace = true }
+tokio = { workspace = true }
+tokio-stream = { workspace = true }
+tokio-util = { workspace = true }
+futures-core = { workspace = true }
+futures-util = { workspace = true }
+thiserror = { workspace = true }
+anyhow = { workspace = true }
+derive_builder = { workspace = true }

crates/download-manager/src/context.rs

@@ -0,0 +1,74 @@
+use reqwest::Client;
+use std::sync::{
+    Arc,
+    atomic::{AtomicU64, AtomicUsize, Ordering},
+};
+use tokio::sync::Semaphore;
+use tokio_util::sync::CancellationToken;
+
+use crate::events::EventBus;
+
+/// Unique identifier for a download; monotonically increasing u64.
+pub type DownloadID = u64;
+
+/// Shared runtime context for coordinating downloads. Internal to the crate.
+/// Holds the concurrency semaphore, root cancellation token, HTTP client,
+/// atomic counters, and the global [DownloadEvent] broadcast sender.
+/// Cloned and shared across scheduler and workers.
+#[derive(Debug)]
+pub(crate) struct Context {
+    /// Semaphore limiting concurrent active downloads.
+    pub semaphore: Arc<Semaphore>,
+    /// Root cancellation token; children inherit via [Context::child_token()].
+    pub cancel_root: CancellationToken,
+    /// Shared reqwest client reused across attempts.
+    pub client: Client,
+
+    // Counters
+    /// Monotonic counter for generating DownloadID values.
+    pub id_counter: AtomicU64,
+    /// Number of currently active (running) downloads.
+    pub active: AtomicUsize,
+    /// Configured maximum concurrency. Not automatically updated if semaphore changes.
+    pub max_concurrent: AtomicUsize,
+
+    /// Global [DownloadEvent] broadcaster (buffered). Slow subscribers may miss events.
+    pub events: EventBus,
+}
+
+impl Context {
+    /// Create a new shared Context.
+    /// - Initializes the semaphore with `max_concurrent` permits.
+    /// - Creates a root [CancellationToken] and a broadcast channel (capacity 1024).
+    /// - Constructs a shared [reqwest::Client].
+    pub fn new(max_concurrent: usize, cancel_root: CancellationToken) -> Arc<Self> {
+        Arc::new(Self {
+            semaphore: Arc::new(Semaphore::new(max_concurrent)),
+            max_concurrent: AtomicUsize::new(max_concurrent),
+            cancel_root,
+            active: AtomicUsize::new(0),
+            id_counter: AtomicU64::new(1),
+            client: Client::new(),
+            events: EventBus::new(),
+        })
+    }
+
+    /// Atomically generate the next [DownloadID] (relaxed ordering).
+    /// Unique within the lifetime of this Context; starts at 1.
+    #[inline]
+    pub fn next_id(&self) -> DownloadID {
+        self.id_counter.fetch_add(1, Ordering::Relaxed)
+    }
+
+    /// Create a child [CancellationToken] tied to the manager's root token.
+    /// Cancelling the root cascades to all children.
+    #[inline]
+    pub fn child_token(&self) -> CancellationToken {
+        self.cancel_root.child_token()
+    }
+
+    /// Cancel the root token, cooperatively cancelling all in-flight downloads.
+    pub fn cancel_all(&self) {
+        self.cancel_root.cancel();
+    }
+}

crates/download-manager/src/download.rs

@@ -0,0 +1,124 @@
+use crate::{DownloadError, DownloadID, Event, Progress};
+use futures_core::Stream;
+use std::path::PathBuf;
+use tokio::sync::{broadcast, oneshot, watch};
+use tokio_stream::wrappers::{BroadcastStream, WatchStream};
+use tokio_util::sync::CancellationToken;
+
+/// Handle for a single download scheduled by DownloadManager.
+///
+/// Behavior:
+/// - Implements Future; awaiting resolves to DownloadResult or DownloadError.
+/// - Exposes per-download streams via [Download::progress()] and [Download::events()].
+/// - Cancellation is cooperative via [Download::cancel()]; the worker aborts the HTTP request and removes any partial file.
+pub struct Download {
+    id: DownloadID,
+    progress: watch::Receiver<Progress>,
+    events: broadcast::Receiver<Event>,
+    result: oneshot::Receiver<Result<DownloadResult, DownloadError>>,
+
+    cancel_token: CancellationToken,
+}
+
+impl Download {
+    pub(crate) fn new(
+        id: DownloadID,
+        progress: watch::Receiver<Progress>,
+        events: broadcast::Receiver<Event>,
+        result: oneshot::Receiver<Result<DownloadResult, DownloadError>>,
+        cancel_token: CancellationToken,
+    ) -> Self {
+        Download {
+            id,
+            progress,
+            events,
+            result,
+            cancel_token,
+        }
+    }
+
+    /// Unique identifier for this download, matching [DownloadEvent] IDs.
+    pub fn id(&self) -> DownloadID {
+        self.id
+    }
+
+    /// Request cooperative cancellation of this download.
+    ///
+    /// The scheduler/worker aborts the in-flight HTTP request and deletes any partially
+    /// written file. Cancellation is best-effort and may race with completion.
+    pub fn cancel(&self) {
+        self.cancel_token.cancel();
+    }
+
+    pub fn progress_raw(&self) -> watch::Receiver<Progress> {
+        self.progress.clone()
+    }
+
+    /// Stream of sampled Progress updates for this download.
+    ///
+    /// Backed by a watch channel: consumers receive the latest state immediately,
+    /// and updates are coalesced according to sampling thresholds.
+    pub fn progress(&self) -> impl Stream<Item = Progress> + 'static {
+        WatchStream::new(self.progress_raw())
+    }
+
+    /// Stream of [DownloadEvent] values scoped to this download only.
+    ///
+    /// Backed by a broadcast channel; lagged consumers may drop messages.
+    /// This stream filters events to those whose id matches this handle.
+    pub fn events(&self) -> impl Stream<Item = Event> + 'static {
+        use tokio_stream::StreamExt as _;
+
+        let download_id = self.id;
+        BroadcastStream::new(self.events.resubscribe())
+            .filter_map(|res| res.ok())
+            .filter(move |event| {
+                let matches = match event {
+                    Event::Queued { id, .. }
+                    | Event::Probed { id, .. }
+                    | Event::Started { id, .. }
+                    | Event::Retrying { id, .. }
+                    | Event::Completed { id, .. }
+                    | Event::Failed { id, .. }
+                    | Event::Cancelled { id, .. } => *id == download_id,
+                };
+
+                matches
+            })
+    }
+}
+
+impl std::future::Future for Download {
+    type Output = Result<DownloadResult, DownloadError>;
+
+    fn poll(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<Self::Output> {
+        use std::pin::Pin;
+        use std::task::Poll;
+
+        match Pin::new(&mut self.result).poll(cx) {
+            Poll::Ready(Ok(result)) => Poll::Ready(result),
+            Poll::Ready(Err(_)) => Poll::Ready(Err(DownloadError::ManagerShutdown)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct DownloadResult {
+    pub path: PathBuf,
+    pub bytes_downloaded: u64,
+}
+
+#[derive(Debug, Clone)]
+/// Remote metadata obtained via a best-effort `HEAD` probe prior to downloading.
+/// Availability depends on server support; fields are None when not provided.
+pub struct RemoteInfo {
+    pub content_length: Option<u64>,
+    pub accept_ranges: Option<String>,
+    pub etag: Option<String>,
+    pub last_modified: Option<String>,
+    pub content_type: Option<String>,
+}