oiupoyt/OneLauncher-Cracked
1
1
Fork 0
Code Issues Pull requests Projects Releases 2 Packages Wiki Activity

fix(core): better parallelisation for downloads and tweaked visual download elements

Browse source
This commit is contained in:
LynithDev 2026-07-25 17:47:46 +02:00
parent 021976ee10
commit b2672c8712
23 changed files with 905 additions and 206 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
packages/oneclient_app/src/bridge/runtime.rs
View file

@ -44,7 +44,7 @@ impl CoreBridgeRuntime {
if let Err(err) = Self::initialize_launcher(&mut snapshots, notifier_tx).await {
let message = err.to_string();
snapshots.launcher.error = Some(message.clone());
let (notif, _timers, _) = notification_engine.dispatch(
notification_engine.dispatch(
&mut inbox,
Notification::Message {
title: "Launcher failed to start".into(),
@ -52,7 +52,13 @@ impl CoreBridgeRuntime {
level: NotificationLevel::Error,
},
);
snapshots.notifications = notif;
sync_notifications(
&mut snapshots,
&notification_engine,
&inbox,
notification_center_open,
&pending_prompt,
);
publish(&self.snapshots_tx, &snapshots);
tracing::error!("launcher init failed: {err:#}");
} else {
@ -159,8 +165,7 @@ impl CoreBridgeRuntime {
immediate = true;
}
other => {
let (notif, _timers, prompt) = engine.dispatch(inbox, other);
snapshots.notifications = notif;
let (_timers, prompt) = engine.dispatch(inbox, other);
if let Some(prompt) = prompt {
*pending = Some(prompt);
}

2
packages/oneclient_app/src/main.rs
View file

@ -32,7 +32,7 @@ impl App for OneClientApp {
fn main() {
let mut builder = Builder::new_multi_thread();
builder.enable_all().max_blocking_threads(16);
builder.enable_all().max_blocking_threads(64);
// Debug builds emit unoptimized async code, which can cause stack overflows in some cases.
// Default stack size is 2MB, so we'll increase it to 4MB for debug builds

28
packages/oneclient_app/src/notifications.rs
View file

@ -171,7 +171,8 @@ pub struct NotificationState {
}
/// Fixed display order for the category rows.
const CATEGORY_ORDER: [TaskCategory; 6] = [
const CATEGORY_ORDER: [TaskCategory; 7] = [
TaskCategory::Java,
TaskCategory::Client,
TaskCategory::Libraries,
TaskCategory::Natives,
@ -293,15 +294,15 @@ impl NotificationState {
inbox.iter().filter(|entry| !entry.read).count()
}
/// Folds one notification into the engine state. Deliberately does not build
/// a [`NotificationSnapshot`]: during a download this runs tens of thousands
/// of times, and every snapshot deep-clones the whole inbox only for the
/// caller to discard all but the last. Callers snapshot once, after draining.
pub fn dispatch(
&mut self,
inbox: &mut Vec<InboxEntry>,
notification: Notification,
) -> (
NotificationSnapshot,
Vec<ToastDismissTimer>,
Option<PendingPrompt>,
) {
) -> (Vec<ToastDismissTimer>, Option<PendingPrompt>) {
self.pending_timers.clear();
match notification {
@ -337,27 +338,18 @@ impl NotificationState {
reply_tx,
} => {
let pending_prompt = Some(PendingPrompt {
title: title.clone(),
question: question.clone(),
kind,
reply_tx: Some(reply_tx),
});
let pending_view = Some(PendingPromptView {
title,
question,
kind,
reply_tx: Some(reply_tx),
});
let timers = std::mem::take(&mut self.pending_timers);
return (
self.snapshot(inbox, false, pending_view),
timers,
pending_prompt,
);
return (timers, pending_prompt);
}
}
let timers = std::mem::take(&mut self.pending_timers);
(self.snapshot(inbox, false, None), timers, None)
(timers, None)
}
pub fn toggle_center(&mut self, _inbox: &mut [InboxEntry], center_open: bool) -> bool {

48
packages/oneclient_app/src/view/onboarding/downloading/mod.rs
View file

@ -5,7 +5,7 @@ use freya::animation::*;
use freya::prelude::*;
use freya::router::RouterContext;
use oneclient_core::notification::{
GroupedProgressEvent, GroupedProgressSession, Notification, NotificationService,
GroupedProgressEvent, GroupedProgressSession, Notification, NotificationService, TaskCategory,
};
use oneclient_core::{
BundleArchive, BundleFile, Cluster, ImportTarget, MigrationSource, SentryExclusion,
@ -61,8 +61,20 @@ struct TaskLine {
phase: &'static str,
current: u64,
total: u64,
category: TaskCategory,
}
/// Display order for the task rows, mirroring the notification centre.
const TASK_LANES: [TaskCategory; 7] = [
TaskCategory::Java,
TaskCategory::Client,
TaskCategory::Libraries,
TaskCategory::Natives,
TaskCategory::Assets,
TaskCategory::Metadata,
TaskCategory::Packages,
];
#[derive(Clone, PartialEq, Default)]
struct GroupedAgg {
children: HashMap<Uuid, TaskLine>,
@ -86,9 +98,37 @@ impl GroupedAgg {
}
}
/// One lane per category, taken round-robin.
///
/// Assets run 32-at-a-time and sort before everything alphabetically, so a
/// flat sorted list showed nothing but assets for the whole game download —
/// libraries, natives and the client were downloading at the same time and
/// never got a row. Round-robin gives every active category a slot before
/// any category gets a second one.
fn task_list(&self) -> Vec<TaskLine> {
let mut tasks: Vec<TaskLine> = self.children.values().cloned().collect();
tasks.sort_by(|a, b| a.label.cmp(&b.label));
let mut lanes: Vec<Vec<TaskLine>> = TASK_LANES
.iter()
.map(|category| {
let mut lane: Vec<TaskLine> = self
.children
.values()
.filter(|task| task.category == *category)
.cloned()
.collect();
lane.sort_by(|a, b| a.label.cmp(&b.label));
lane
})
.collect();
let deepest = lanes.iter().map(Vec::len).max().unwrap_or(0);
let mut tasks = Vec::new();
for index in 0..deepest {
for lane in &mut lanes {
if let Some(task) = lane.get(index) {
tasks.push(task.clone());
}
}
}
tasks
}
@ -822,6 +862,7 @@ fn apply_grouped(agg: &mut GroupedAgg, event: GroupedProgressEvent) {
child_id,
label,
total,
category,
..
} => {
agg.children.insert(
@ -831,6 +872,7 @@ fn apply_grouped(agg: &mut GroupedAgg, event: GroupedProgressEvent) {
phase: "Downloading",
current: 0,
total: total.max(1),
category,
},
);
}

2
packages/oneclient_core/examples/java_install_provider.rs
View file

@ -51,7 +51,7 @@ async fn main() -> LauncherResult<()> {
println!("Installing package {:#?} to '{:?}'", package, java_dir()?);
let executable = provider.install_package(&package, &services).await?;
let executable = provider.install_package(&package, &services, None).await?;
println!("Installed: {}", executable.display());

4
packages/oneclient_core/src/bundles/install.rs
View file

@ -282,7 +282,9 @@ pub async fn install_enabled_bundle_files(
Ok(installed)
}
const BUNDLE_INSTALL_CONCURRENCY: usize = 6;
/// How many bundle packages are fetched at once. Kept modest because each one
/// also costs a provider API call, and those are rate limited per-minute.
pub(crate) const BUNDLE_INSTALL_CONCURRENCY: usize = 6;
#[tracing::instrument(level = "debug", skip(bundles, services))]
pub async fn enabled_bundle_bytes(

3
packages/oneclient_core/src/bundles/manager.rs
View file

@ -366,12 +366,13 @@ async fn download_bundle_if_needed(
polyio::create_dir_all(parent).await?;
}
let size_hint = res.content_length();
let stream = res
.stream(ResponseOptions::default(), &services.notifier)
.await
.map_err(map_request_error)?;
let stream = std::pin::pin!(stream);
polyio::write_stream(disk_path, stream)
polyio::write_stream(disk_path, stream, size_hint)
.await
.map_err(map_request_error)?;

156
packages/oneclient_core/src/bundles/updates.rs
View file

@ -9,7 +9,11 @@ use oneclient_db::models::ClusterPatch;
use oneclient_db::models::{BundleTrackedArtifactRow, ClusterBundleOverrideRow, OverrideType};
use tokio::sync::Mutex as AsyncMutex;
use crate::bundles::install::{install_package_from_bundle, remove_artifact_from_cluster};
use futures_util::StreamExt;
use crate::bundles::install::{
BUNDLE_INSTALL_CONCURRENCY, install_package_from_bundle, remove_artifact_from_cluster,
};
use crate::bundles::manager::BundlesManager;
use crate::bundles::overrides;
use crate::bundles::types::{
@ -403,35 +407,94 @@ pub async fn apply_bundle_updates_with(
s.expect(crate::notification::TaskCategory::Packages, count, bytes);
}
for update in check.updates_available {
let child = session.map(|s| {
let c = s.child(
update.new_file.display_name(),
update.new_file.size.max(1),
crate::notification::TaskCategory::Packages,
);
c.set_phase(crate::notification::TaskPhase::Downloading);
c
});
match apply_single_update(&update, &overrides, child.as_ref(), services).await {
Ok(_) => result.updates_applied.push(update),
// Fetch concurrently, reconcile in order. The install path already fans out
// over its files; applying updates one at a time made the same work take
// several times longer here than it does during a fresh install.
//
// Splitting fetch from reconcile is what makes the fan-out safe: no artifact
// is unlinked while another package is still downloading, so an update that
// supersedes a hash another update is installing can't race it.
let fetched_updates = futures_util::stream::iter(check.updates_available.into_iter().map(
|update| async move {
let child = session.map(|s| {
let c = s.child(
update.new_file.display_name(),
update.new_file.size.max(1),
crate::notification::TaskCategory::Packages,
);
c.set_phase(crate::notification::TaskPhase::Downloading);
c
});
let installed = install_package_from_bundle(
&update.new_file,
update.cluster_id,
&update.bundle_name,
true,
child.as_ref(),
services,
)
.await;
if let Some(child) = child {
child.set_phase(crate::notification::TaskPhase::Installing);
child.finish();
}
(update, installed)
},
))
.buffer_unordered(BUNDLE_INSTALL_CONCURRENCY)
.collect::<Vec<_>>()
.await;
for (update, installed) in fetched_updates {
match installed {
Ok(hash) => match reconcile_update(&update, &hash, &overrides, services).await {
Ok(()) => result.updates_applied.push(update),
Err(err) => result.updates_failed.push(err.to_string()),
},
Err(err) => result.updates_failed.push(err.to_string()),
}
}
for addition in check.additions_available {
let fetched_additions = futures_util::stream::iter(check.additions_available.into_iter().map(
|addition| async move {
let child = session.map(|s| {
let c = s.child(
addition.new_file.display_name(),
addition.new_file.size.max(1),
crate::notification::TaskCategory::Packages,
);
c.set_phase(crate::notification::TaskPhase::Downloading);
c
});
let installed = install_package_from_bundle(
&addition.new_file,
addition.cluster_id,
&addition.bundle_name,
true,
child.as_ref(),
services,
)
.await;
if let Some(child) = child {
child.set_phase(crate::notification::TaskPhase::Installing);
child.finish();
}
(addition, installed)
},
))
.buffer_unordered(BUNDLE_INSTALL_CONCURRENCY)
.collect::<Vec<_>>()
.await;
for (addition, installed) in fetched_additions {
let file_id = addition.new_file.kind.package_id();
let child = session.map(|s| {
let c = s.child(
addition.new_file.display_name(),
addition.new_file.size.max(1),
crate::notification::TaskCategory::Packages,
);
c.set_phase(crate::notification::TaskPhase::Downloading);
c
});
match apply_single_addition(&addition, &overrides, child.as_ref(), services).await {
Ok(_) => result.additions_applied.push(addition),
match installed {
Ok(hash) => {
match reconcile_addition(&addition, &hash, &overrides, services).await {
Ok(()) => result.additions_applied.push(addition),
Err(err) => result.additions_failed.push(format!("{file_id}: {err:#}")),
}
}
Err(err) => result.additions_failed.push(format!("{file_id}: {err:#}")),
}
}
@ -471,30 +534,19 @@ pub async fn apply_bundle_updates_with(
Ok(result)
}
/// Local bookkeeping for an update whose new file is already downloaded and
/// linked: apply the user's enable/disable choice, then drop the artifact the
/// update supersedes.
#[tracing::instrument(level = "debug", skip_all, fields(cluster_id = update.cluster_id, bundle = %update.bundle_name, new_version = %update.new_version_id))]
async fn apply_single_update(
async fn reconcile_update(
update: &BundlePackageUpdate,
hash: &str,
overrides: &[ClusterBundleOverrideRow],
child: Option<&crate::notification::GroupedProgressChild>,
services: &LauncherServices,
) -> LauncherResult<()> {
let hash = install_package_from_bundle(
&update.new_file,
update.cluster_id,
&update.bundle_name,
true,
child,
services,
)
.await?;
if let Some(child) = child {
child.set_phase(crate::notification::TaskPhase::Installing);
child.finish();
}
let file_id = update.new_file.kind.package_id();
if should_be_disabled(&update.bundle_name, &file_id, overrides) {
PackageStore::toggle_artifact_enabled(update.cluster_id, &hash, services).await?;
PackageStore::toggle_artifact_enabled(update.cluster_id, hash, services).await?;
}
if hash == update.installed_hash {
@ -505,29 +557,15 @@ async fn apply_single_update(
}
#[tracing::instrument(level = "debug", skip_all, fields(cluster_id = addition.cluster_id, bundle = %addition.bundle_name))]
async fn apply_single_addition(
async fn reconcile_addition(
addition: &BundlePackageAddition,
hash: &str,
overrides: &[ClusterBundleOverrideRow],
child: Option<&crate::notification::GroupedProgressChild>,
services: &LauncherServices,
) -> LauncherResult<()> {
let hash = install_package_from_bundle(
&addition.new_file,
addition.cluster_id,
&addition.bundle_name,
true,
child,
services,
)
.await?;
if let Some(child) = child {
child.set_phase(crate::notification::TaskPhase::Installing);
child.finish();
}
let file_id = addition.new_file.kind.package_id();
if should_be_disabled(&addition.bundle_name, &file_id, overrides) {
PackageStore::toggle_artifact_enabled(addition.cluster_id, &hash, services).await?;
PackageStore::toggle_artifact_enabled(addition.cluster_id, hash, services).await?;
}
Ok(())

58
packages/oneclient_core/src/clusters/prepare.rs
View file

@ -17,7 +17,7 @@ use crate::java::JavaManager;
use crate::metadata::MetadataStore;
use crate::notification::GroupedProgressSession;
use crate::paths;
use crate::state::LauncherState;
use crate::state::{LauncherServices, LauncherState};
use crate::{GameError, LauncherResult};
#[tracing::instrument(skip(state, metadata, shared_progress))]
@ -83,21 +83,44 @@ pub async fn prepare_cluster(
const JRE_ESTIMATE_BYTES: u64 = 45_000_000;
fn game_download_bytes(info: &VersionInfo) -> u64 {
let client = info
.downloads
.get(&DownloadType::Client)
.map(|d| d.size as u64)
.unwrap_or(0);
let assets = info.asset_index.total_size as u64;
let libraries: u64 = info
.libraries
.iter()
.filter_map(|lib| lib.downloads.as_ref())
.filter_map(|dl| dl.artifact.as_ref())
.map(|artifact| artifact.size as u64)
.sum();
client + assets + libraries
/// Bytes the game install still needs. Falls back to the manifest's full size
/// when the asset index isn't cached yet — which is exactly the case where
/// nothing is on disk, so the full size is the right answer anyway.
async fn game_download_bytes(services: &LauncherServices, info: &VersionInfo) -> u64 {
let Some(assets_index) = cached_assets_index(info).await else {
let client = info
.downloads
.get(&DownloadType::Client)
.map_or(0, |d| d.size as u64);
let libraries: u64 = info
.libraries
.iter()
.filter_map(|lib| lib.downloads.as_ref())
.filter_map(|dl| dl.artifact.as_ref())
.map(|artifact| artifact.size as u64)
.sum();
return client + info.asset_index.total_size as u64 + libraries;
};
let _ = services;
// The estimate runs before a Java runtime is resolved, so rules are checked
// against the host architecture. That only shifts natives-related edge cases.
match game::plan_downloads(info, assets_index, std::env::consts::ARCH, false, false).await {
Ok(plan) => plan.total_bytes(),
Err(err) => {
tracing::warn!("download plan failed, estimating full size: {err}");
info.asset_index.total_size as u64
}
}
}
async fn cached_assets_index(
info: &VersionInfo,
) -> Option<interfrost::api::minecraft::AssetsIndex> {
let path = paths::assets_index_dir()
.ok()?
.join(format!("{}.json", info.asset_index.id));
polyio::read_json(&path).await.ok()
}
#[tracing::instrument(level = "debug", skip(state, bundles))]
@ -126,7 +149,7 @@ pub async fn estimate_cluster_download(
download_version_info(&state.services, None, &version, loader_version.as_ref(), false).await?
};
let mut total = game_download_bytes(&info);
let mut total = game_download_bytes(&state.services, &info).await;
if let Some(java) = &info.java_version {
let installed = JavaManager::list_runtimes(&state.services.db)
@ -197,6 +220,7 @@ async fn install_cluster(
java_major,
search_for_java,
auto_install_java,
Some(progress),
)
.await?
};

69
packages/oneclient_core/src/crypto.rs
View file

@ -1,32 +1,77 @@
use std::fmt::Write;
use std::io::Read;
use digest::Digest;
use tokio::io::AsyncReadExt;
use crate::LauncherResult;
/// Largest read chunk used when hashing a file. Files smaller than this get a
/// buffer sized to fit, so hashing a 10 KiB asset doesn't allocate (and zero)
/// a quarter of a megabyte.
const MAX_HASH_BUFFER: usize = 256 * 1024;
/// Incremental SHA-1, so a download can be verified from the bytes as they
/// stream past instead of re-reading the finished file back off disk.
pub struct Sha1Stream(sha1::Sha1);
impl Sha1Stream {
#[must_use]
pub fn new() -> Self {
Self(sha1::Sha1::new())
}
pub fn update(&mut self, data: &[u8]) {
Digest::update(&mut self.0, data);
}
#[must_use]
pub fn finish(self) -> String {
to_hex(&self.0.finalize())
}
}
impl Default for Sha1Stream {
fn default() -> Self {
Self::new()
}
}
/// Hashing is CPU- and syscall-bound, so it runs on the blocking pool: one
/// dispatch for the whole file rather than one per read, and the digest itself
/// stays off the async workers that are driving concurrent downloads.
pub async fn sha1_file(path: impl AsRef<std::path::Path>) -> LauncherResult<String> {
let path = path.as_ref();
let mut hasher = sha1::Sha1::new();
let file = tokio::fs::File::open(path).await?;
let mut reader = tokio::io::BufReader::new(file);
let mut buffer = vec![0u8; 256 * 1024];
let path = path.as_ref().to_path_buf();
tokio::task::spawn_blocking(move || sha1_file_sync(&path))
.await
.map_err(std::io::Error::other)?
}
pub fn sha1_file_sync(path: &std::path::Path) -> LauncherResult<String> {
let mut file = std::fs::File::open(path)?;
let mut hasher = Sha1Stream::new();
let size = file
.metadata()
.map(|meta| meta.len())
.unwrap_or(MAX_HASH_BUFFER as u64);
let capacity = (size.max(1).min(MAX_HASH_BUFFER as u64)) as usize;
let mut buffer = vec![0u8; capacity];
loop {
let n = reader.read(&mut buffer).await?;
let n = file.read(&mut buffer)?;
if n == 0 {
break;
}
Digest::update(&mut hasher, &buffer[..n]);
hasher.update(&buffer[..n]);
}
Ok(to_hex(&hasher.finalize()))
Ok(hasher.finish())
}
pub fn sha1_bytes(data: &[u8]) -> String {
let mut hasher = sha1::Sha1::new();
Digest::update(&mut hasher, data);
to_hex(&hasher.finalize())
let mut hasher = Sha1Stream::new();
hasher.update(data);
hasher.finish()
}
pub fn normalize_hash(hash: &str) -> String {

51
packages/oneclient_core/src/game/download.rs
View file

@ -1,6 +1,6 @@
use reqwest::Method;
use crate::crypto::{normalize_hash, sha1_bytes, sha1_file};
use crate::crypto::{Sha1Stream, normalize_hash, sha1_bytes};
use crate::game::GameError;
use crate::http::{RequestClient, ResponseExt, ResponseNotifyOptions, ResponseOptions};
use crate::notification::{GroupedProgressSession, NotificationService, TaskCategory, TaskPhase};
@ -19,13 +19,25 @@ pub async fn download_to_path(
dest: &std::path::Path,
expected_sha1: Option<&str>,
) -> LauncherResult<()> {
if let Some(parent) = dest.parent() {
// Callers that fan out over thousands of files (assets) pre-create the
// directory tree, so this is usually a stat that avoids a blocking-pool
// round trip per file.
if let Some(parent) = dest.parent()
&& !parent.is_dir()
{
polyio::create_dir_all(parent).await?;
}
let label = label.into();
let parsed_url = url.parse().map_err(LauncherError::UrlError)?;
let request = reqwest::Request::new(Method::GET, parsed_url);
// Created before the request goes out: waiting on response headers is most
// of a small file's lifetime, and a child that only exists while bytes are
// moving never appears in the UI at all. `set_progress` raises the total if
// Content-Length turns out to be larger than the manifest size.
let child = progress.child(label, expected_size, category);
let response = requester.send(request).await?;
// Prefer the manifest size; fall back to Content-Length only when unknown.
@ -34,7 +46,6 @@ pub async fn download_to_path(
} else {
response.content_length().unwrap_or(0).max(1)
};
let child = progress.child(label, total, category);
let stream = response
.stream(
@ -44,13 +55,25 @@ pub async fn download_to_path(
notifier,
)
.await?;
let stream = std::pin::pin!(stream);
polyio::write_stream(dest, stream).await?;
if let Some(expected) = expected_sha1 {
child.set_phase(TaskPhase::Verifying);
let actual = sha1_file(dest).await?;
if normalize_hash(&actual) != normalize_hash(expected) {
// Hash the bytes as they stream past. Verifying by re-reading the finished
// file costs a second full pass over every download, and it does it while
// still holding a slot in the caller's concurrency fan-out.
let mut hasher = expected_sha1.map(|_| Sha1Stream::new());
{
let stream = futures_lite::StreamExt::map(stream, |item| {
if let (Ok(chunk), Some(hasher)) = (&item, hasher.as_mut()) {
hasher.update(chunk);
}
item
});
let stream = std::pin::pin!(stream);
polyio::write_stream(dest, stream, Some(total)).await?;
}
if let (Some(expected), Some(hasher)) = (expected_sha1, hasher) {
let actual = hasher.finish();
if actual != normalize_hash(expected) {
let _ = polyio::remove_file(dest).await;
return Err(GameError::HashMismatch {
path: dest.display().to_string(),
@ -79,14 +102,10 @@ pub async fn fetch_bytes_verified(
let label = label.into();
let parsed_url: url::Url = url.parse().map_err(LauncherError::UrlError)?;
let request = reqwest::Request::new(Method::GET, parsed_url.clone());
let response = requester.send(request).await?;
let total = if expected_size > 0 {
expected_size
} else {
response.content_length().unwrap_or(0).max(1)
};
let child = progress.child(label, total, category);
let child = progress.child(label, expected_size, category);
let response = requester.send(request).await?;
let stream = response
.stream(

266
packages/oneclient_core/src/game/metadata.rs
View file

@ -1,7 +1,9 @@
use std::path::PathBuf;
use std::path::{Path, PathBuf};
use futures_util::{StreamExt, stream};
use interfrost::api::minecraft::{AssetsIndex, DownloadType, Library, Os, Version, VersionInfo};
use interfrost::api::minecraft::{
Asset, AssetsIndex, DownloadType, Library, Os, Version, VersionInfo,
};
use interfrost::api::modded::LoaderVersion;
use reqwest::Method;
@ -17,6 +19,146 @@ use crate::paths;
use crate::state::LauncherServices;
use crate::{LauncherError, LauncherResult};
/// Asset objects are tiny (median ~10 KiB) and latency-bound, so throughput
/// scales with how many are in flight, not with bandwidth.
const ASSET_DOWNLOAD_CONCURRENCY: usize = 32;
/// Libraries are larger and fewer; less fan-out is needed to saturate the link.
const LIBRARY_DOWNLOAD_CONCURRENCY: usize = 16;
/// Everything a version still needs, with what is already on disk subtracted.
///
/// Computed once, up-front, so the size shown before a download starts and the
/// denominators the progress bars fill against come from the same numbers —
/// previously the estimate assumed a full download while the bars discovered
/// their own total as files were added, so neither matched reality.
#[derive(Debug, Default)]
pub struct DownloadPlan {
pub assets: Vec<(String, Asset)>,
pub asset_bytes: u64,
pub libraries: Vec<Library>,
pub library_bytes: u64,
pub client: bool,
pub client_bytes: u64,
}
impl DownloadPlan {
#[must_use]
pub fn total_bytes(&self) -> u64 {
self.asset_bytes + self.library_bytes + self.client_bytes
}
}
fn asset_object_path(dir: &Path, name: &str, hash: &str, legacy: bool) -> PathBuf {
if legacy {
dir.join(name.replace('/', std::path::MAIN_SEPARATOR_STR))
} else {
dir.join(&hash[0..2]).join(hash)
}
}
fn native_download_size(lib: &Library, java_arch: &str) -> u64 {
let Some((os_key, classifiers)) = lib.natives.as_ref().and_then(|natives| {
Some((
natives.get(&Os::native_arch(java_arch))?,
lib.downloads.as_ref()?.classifiers.as_ref()?,
))
}) else {
return 0;
};
let parsed = os_key.replace("${arch}", crate::constants::ARCH_WIDTH);
classifiers
.get(&parsed)
.map_or(0, |native| u64::from(native.size))
}
/// Stats every candidate file, so it runs on the blocking pool — an asset index
/// is thousands of entries. Doing it here also means the download fan-outs no
/// longer stat each file from inside a concurrency slot.
#[tracing::instrument(skip_all, level = "debug")]
pub async fn plan_downloads(
version: &VersionInfo,
assets_index: AssetsIndex,
java_arch: &str,
minecraft_updated: bool,
force: bool,
) -> LauncherResult<DownloadPlan> {
let legacy = version.assets == "legacy";
let asset_dir = if legacy {
paths::legacy_assets_dir()?
} else {
paths::assets_object_dir()?
};
let lib_dir = paths::libraries_dir()?;
let client = version.downloads.get(&DownloadType::Client).map(|client| {
(
paths::versions_dir()
.map(|dir| dir.join(&version.id).join(format!("{}.jar", version.id))),
u64::from(client.size),
)
});
let libraries = version.libraries.clone();
let java_arch = java_arch.to_string();
tokio::task::spawn_blocking(move || {
let mut plan = DownloadPlan::default();
for (name, asset) in assets_index.objects {
let path = asset_object_path(&asset_dir, &name, &asset.hash, legacy);
if !force && path.exists() {
continue;
}
plan.asset_bytes += u64::from(asset.size);
plan.assets.push((name, asset));
}
for lib in libraries {
if let Some(rules) = &lib.rules
&& !validate_rules(rules, &java_arch, minecraft_updated)
{
continue;
}
if !lib.downloadable {
continue;
}
// A library whose coordinates don't resolve is kept in the plan so the
// download path still reports it as a failure rather than skipping it.
let Ok(artifact_path) = interfrost::utils::get_path_from_artifact(&lib.name) else {
plan.libraries.push(lib);
continue;
};
if !force && lib_dir.join(&artifact_path).exists() {
continue;
}
plan.library_bytes += lib
.downloads
.as_ref()
.and_then(|downloads| downloads.artifact.as_ref())
.map_or(0, |artifact| u64::from(artifact.size));
plan.library_bytes += native_download_size(&lib, &java_arch);
plan.libraries.push(lib);
}
if let Some((path, size)) = client {
let present = path.map(|path| path.exists()).unwrap_or(false);
if force || !present {
plan.client = true;
plan.client_bytes = size;
}
}
plan
})
.await
.map_err(std::io::Error::other)
.map_err(Into::into)
}
#[tracing::instrument(skip_all)]
pub async fn download_minecraft(
services: &LauncherServices,
@ -26,27 +168,59 @@ pub async fn download_minecraft(
minecraft_updated: bool,
force: bool,
) -> LauncherResult<()> {
let started = std::time::Instant::now();
let asset_index = download_assets_index(services, progress, version, force).await?;
let plan = plan_downloads(version, asset_index, java_arch, minecraft_updated, force).await?;
tracing::info!(
version = %version.id,
assets = plan.assets.len(),
libraries = plan.libraries.len(),
bytes = plan.total_bytes(),
"planned minecraft download"
);
// Reserve each category's real total before any child appears, so the bars
// start at the right denominator instead of growing towards it.
progress.expect(
TaskCategory::Assets,
plan.assets.len() as u64,
plan.asset_bytes,
);
progress.expect(
TaskCategory::Libraries,
plan.libraries.len() as u64,
plan.library_bytes,
);
progress.expect(TaskCategory::Client, u64::from(plan.client), plan.client_bytes);
let DownloadPlan {
assets, libraries, ..
} = plan;
tokio::try_join!(
download_assets(
services,
progress,
version.assets == "legacy",
asset_index,
force
assets,
),
download_client(services, progress, version, force),
download_libraries(
services,
progress,
version.id.clone(),
version.libraries.clone(),
libraries,
java_arch,
minecraft_updated,
force,
),
)?;
tracing::info!(
version = %version.id,
elapsed_ms = started.elapsed().as_millis() as u64,
"minecraft download complete"
);
Ok(())
}
@ -194,9 +368,12 @@ pub async fn download_assets(
services: &LauncherServices,
progress: &GroupedProgressSession,
legacy: bool,
assets_index: AssetsIndex,
force: bool,
assets: Vec<(String, Asset)>,
) -> LauncherResult<usize> {
if assets.is_empty() {
return Ok(0);
}
let dir = if legacy {
paths::legacy_assets_dir()?
} else {
@ -205,10 +382,25 @@ pub async fn download_assets(
polyio::create_dir_all(&dir).await?;
// Objects are sharded into 256 fixed subdirectories; creating them once
// here saves a blocking-pool round trip on every one of the ~5000 files.
if !legacy {
let dir = dir.clone();
tokio::task::spawn_blocking(move || {
for subhash in 0u16..256 {
let _ = std::fs::create_dir_all(dir.join(format!("{subhash:02x}")));
}
})
.await
.map_err(std::io::Error::other)?;
}
let started = std::time::Instant::now();
let count = assets.len();
let requester = services.requester.clone();
let notifier = services.notifier.clone();
let progress = progress.clone();
let requests = stream::iter(assets_index.objects.into_iter().map(|(name, asset)| {
let requests = stream::iter(assets.into_iter().map(|(name, asset)| {
let dir = dir.clone();
let requester = requester.clone();
let notifier = notifier.clone();
@ -217,15 +409,7 @@ pub async fn download_assets(
async move {
let hash = &asset.hash;
let subhash = &hash[0..2];
let path = if legacy {
dir.join(name.replace('/', std::path::MAIN_SEPARATOR_STR))
} else {
dir.join(subhash).join(hash)
};
if path.exists() && !force {
return Ok::<(), LauncherError>(());
}
let path = asset_object_path(&dir, &name, hash, legacy);
let url = format!("https://resources.download.minecraft.net/{subhash}/{hash}");
download_to_path(
@ -242,7 +426,7 @@ pub async fn download_assets(
.await
}
}))
.buffer_unordered(7)
.buffer_unordered(ASSET_DOWNLOAD_CONCURRENCY)
.collect::<Vec<_>>();
let mut failed = 0;
@ -253,6 +437,13 @@ pub async fn download_assets(
}
}
tracing::info!(
count,
failed,
elapsed_ms = started.elapsed().as_millis() as u64,
"assets ready"
);
Ok(failed)
}
@ -339,9 +530,11 @@ pub async fn download_libraries(
version: String,
libraries: Vec<Library>,
java_arch: &str,
minecraft_updated: bool,
force: bool,
) -> LauncherResult<usize> {
if libraries.is_empty() {
return Ok(0);
}
let lib_dir = paths::libraries_dir()?;
let natives_dest = paths::natives_dir()?.join(&version);
let java_arch = java_arch.to_string();
@ -349,6 +542,8 @@ pub async fn download_libraries(
polyio::create_dir_all(&lib_dir).await?;
polyio::create_dir_all(&natives_dest).await?;
let started = std::time::Instant::now();
let count = libraries.len();
let requests = stream::iter(libraries.into_iter().map(|lib| {
let lib_dir = lib_dir.clone();
let natives_dest = natives_dest.clone();
@ -358,24 +553,12 @@ pub async fn download_libraries(
let progress = progress.clone();
async move {
if let Some(rules) = &lib.rules
&& !validate_rules(rules, &java_arch, minecraft_updated)
{
return Ok::<(), LauncherError>(());
}
if !lib.downloadable {
return Ok(());
}
// Rule evaluation and the on-disk check already happened in
// `plan_downloads`; everything here is known to need fetching.
let artifact_path = interfrost::utils::get_path_from_artifact(&lib.name)
.map_err(|_| GameError::LibraryPath(lib.name.clone()))?;
let path = lib_dir.join(&artifact_path);
if path.exists() && !force {
return Ok(());
}
tokio::try_join!(
async {
if let Some(interfrost::api::minecraft::LibraryDownloads {
@ -461,10 +644,10 @@ pub async fn download_libraries(
}
)?;
Ok(())
Ok::<(), LauncherError>(())
}
}))
.buffer_unordered(7)
.buffer_unordered(LIBRARY_DOWNLOAD_CONCURRENCY)
.collect::<Vec<_>>();
let mut failed = 0;
@ -475,6 +658,13 @@ pub async fn download_libraries(
}
}
tracing::info!(
count,
failed,
elapsed_ms = started.elapsed().as_millis() as u64,
"libraries ready"
);
Ok(failed)
}

2
packages/oneclient_core/src/game/mod.rs
View file

@ -28,7 +28,7 @@ pub use process::{
};
pub use reattach::recover_sessions;
pub use metadata::{
download_minecraft, download_version_info, get_game_versions, get_loader_version,
DownloadPlan, download_minecraft, plan_downloads, download_version_info, get_game_versions, get_loader_version,
get_loader_versions, get_loaders_for_version, is_version_updated, libraries_missing,
resolve_minecraft_version,
};

35
packages/oneclient_core/src/http/response.rs
View file

@ -4,6 +4,9 @@ use uuid::Uuid;
use crate::http::RequestError;
use crate::notification::{GroupedProgressChild, NotificationService};
/// Minimum gap between progress events emitted for a single response body.
const PROGRESS_INTERVAL: std::time::Duration = std::time::Duration::from_millis(50);
#[derive(Clone, Default)]
pub struct ResponseOptions {
pub notify: Option<ResponseNotifyOptions>,
@ -92,21 +95,33 @@ impl ResponseExt for Response {
}
let notifier = notifier.clone();
// A chunk arrives every few KiB; forwarding one event per chunk buries the
// UI in tens of thousands of updates per download session, all of which
// collapse into the same repainted progress bar. Sample instead, and
// always emit the last one so the bar lands on complete.
let mut last_emit: Option<std::time::Instant> = None;
let stream = futures_lite::StreamExt::map(self.bytes_stream(), move |item| {
match item {
Ok(chunk) => {
current += chunk.len() as u64;
if let Some(ref child) = grouped_child {
child.set_progress(current, Some(total));
} else if let (Some(id), Some(label)) = (&standalone_id, &standalone_label) {
let done = current >= total;
let label = if done {
done_label.as_deref().unwrap_or(label)
} else {
label
};
notifier.send_progress(id, label, current, total);
let now = std::time::Instant::now();
let due = current >= total
|| last_emit.is_none_or(|last| now.duration_since(last) >= PROGRESS_INTERVAL);
if due {
last_emit = Some(now);
if let Some(ref child) = grouped_child {
child.set_progress(current, Some(total));
} else if let (Some(id), Some(label)) = (&standalone_id, &standalone_label) {
let done = current >= total;
let label = if done {
done_label.as_deref().unwrap_or(label)
} else {
label
};
notifier.send_progress(id, label, current, total);
}
}
Ok(chunk)

12
packages/oneclient_core/src/http/service.rs
View file

@ -8,6 +8,13 @@ use crate::{LauncherServices, http::{HttpRequest, RequestError, ResponseExt, Res
const MAX_THROTTLE_RETRIES: u32 = 6;
/// Ceiling on requests waiting for response headers. This is a backstop against
/// a runaway fan-out, not the download throttle — per-phase concurrency is set
/// by the callers, and this has to stay above their sum or it becomes the
/// bottleneck instead. Permits are released once headers arrive, so streaming
/// bodies don't hold a slot.
const MAX_INFLIGHT_REQUESTS: usize = 64;
fn retry_after(response: &Response) -> Option<std::time::Duration> {
let raw = response
.headers()
@ -123,7 +130,7 @@ impl RequestClient {
Ok(Self {
client,
semaphore: Arc::new(Semaphore::new(12)),
semaphore: Arc::new(Semaphore::new(MAX_INFLIGHT_REQUESTS)),
})
}
}
@ -219,10 +226,11 @@ impl RequestClient {
services: &LauncherServices,
) -> Result<(), RequestError> {
let res = self.send(request).await?;
let size_hint = res.content_length();
let http_stream = res.stream(options, &services.notifier).await?;
let http_stream = std::pin::pin!(http_stream);
polyio::write_stream(dest, http_stream).await?;
polyio::write_stream(dest, http_stream, size_hint).await?;
Ok(())
}

34
packages/oneclient_core/src/java/install.rs
View file

@ -7,14 +7,16 @@ use crate::http::{ResponseNotifyOptions, ResponseOptions};
use crate::java::data::{
java_executable_relative_path, JavaPackage, PackageArchive,
};
use crate::notification::{GroupedProgressSession, TaskCategory, TaskPhase};
use crate::paths;
use crate::state::LauncherServices;
use crate::LauncherResult;
#[tracing::instrument(level = "debug", skip(services))]
#[tracing::instrument(level = "debug", skip(services, progress))]
pub async fn install_package(
package: &JavaPackage,
services: &LauncherServices,
progress: Option<&GroupedProgressSession>,
) -> LauncherResult<PathBuf> {
let java_dir = paths::java_dir()?;
polyio::create_dir_all(&java_dir).await?;
@ -22,18 +24,30 @@ pub async fn install_package(
let archive_path = java_dir.join(polyio::sanitize_path(&package.name));
let major = package.java_version.first().copied().unwrap_or(0);
// When a caller owns a grouped session (onboarding, cluster prepare) the
// runtime is one of that session's tasks. Only a standalone install — from
// settings, say — gets its own notification.
// No size in the package metadata; the child picks up its real total from
// Content-Length as soon as the response headers land.
let child = progress
.map(|session| session.child(format!("{} {}", package.vendor, major), 0, TaskCategory::Java));
let notify = match &child {
Some(child) => ResponseNotifyOptions::grouped(child.clone()),
None => ResponseNotifyOptions::standalone(format!(
"Installing {} {}",
package.vendor, major
))
.done_label(format!("Installed {} {}", package.vendor, major)),
};
services.requester
.download_file(
reqwest::Request::new(Method::GET, Url::parse(&package.download_url)?),
&archive_path,
ResponseOptions {
notify: Some(
ResponseNotifyOptions::standalone(format!(
"Installing {} {}",
package.vendor, major
))
.done_label(format!("Installed {} {}", package.vendor, major)),
),
notify: Some(notify),
},
services,
)
@ -41,6 +55,10 @@ pub async fn install_package(
let extract_root = java_dir.join(stem_without_archive(&package.name));
if let Some(child) = &child {
child.set_phase(TaskPhase::Extracting);
}
match package.archive {
PackageArchive::Zip => polyio::extract_zip(&archive_path, &extract_root).await?,
PackageArchive::TarGz => polyio::extract_tar_gz(&archive_path, &extract_root).await?,

20
packages/oneclient_core/src/java/manager.rs
View file

@ -30,7 +30,7 @@ impl JavaManager {
major: u32,
search_system: bool,
) -> LauncherResult<JavaRuntime> {
Self::prepare_java_with_services(&state.services, major, search_system, false).await
Self::prepare_java_with_services(&state.services, major, search_system, false, None).await
}
#[tracing::instrument(skip(state))]
@ -39,15 +39,16 @@ impl JavaManager {
major: u32,
search_system: bool,
) -> LauncherResult<JavaRuntime> {
Self::prepare_java_with_services(&state.services, major, search_system, true).await
Self::prepare_java_with_services(&state.services, major, search_system, true, None).await
}
#[tracing::instrument(level = "debug", skip(services))]
#[tracing::instrument(level = "debug", skip(services, progress))]
pub async fn prepare_java_with_services(
services: &LauncherServices,
major: u32,
search_system: bool,
auto_install: bool,
progress: Option<&crate::notification::GroupedProgressSession>,
) -> LauncherResult<JavaRuntime> {
if let Some(runtime) = get_latest_runtime(&services.db, major).await? {
return Ok(runtime);
@ -62,11 +63,11 @@ impl JavaManager {
}
if auto_install {
return download_and_register(major, services).await;
return download_and_register(major, services, progress).await;
}
match services.notifier.prompt_java_install(major).await? {
UserChoice::Accept => download_and_register(major, services).await,
UserChoice::Accept => download_and_register(major, services, progress).await,
UserChoice::Folder(folder) => register_custom_java(&services.db, folder, major).await,
UserChoice::Install { vendor, .. } => {
Self::install_runtime_from(services, &vendor, major).await
@ -91,7 +92,7 @@ impl JavaManager {
services: &LauncherServices,
major: u32,
) -> LauncherResult<JavaRuntime> {
download_and_register(major, services).await
download_and_register(major, services, None).await
}
#[tracing::instrument(level = "debug", skip(services))]
@ -137,7 +138,7 @@ impl JavaManager {
.latest_package_by_major(major, services)
.await?
.ok_or(JavaError::PackageNotFound { major })?;
let executable = provider.install_package(&package, services).await?;
let executable = provider.install_package(&package, services, None).await?;
register_checked_java(&services.db, &executable, Some(major)).await
}
@ -180,10 +181,11 @@ fn provider_for_vendor(vendor: &JavaVendor) -> Option<Box<dyn vendors::JavaRunti
.find(|provider| &provider.vendor() == vendor)
}
#[tracing::instrument(level = "debug", skip(services))]
#[tracing::instrument(level = "debug", skip(services, progress))]
async fn download_and_register(
major: u32,
services: &LauncherServices,
progress: Option<&crate::notification::GroupedProgressSession>,
) -> LauncherResult<JavaRuntime> {
for provider in vendors::runtime_providers() {
let vendor = provider.vendor();
@ -206,7 +208,7 @@ async fn download_and_register(
tracing::info!(?vendor, major, "downloading Java runtime");
match provider.install_package(package, services).await {
match provider.install_package(package, services, progress).await {
Ok(executable) => {
return register_checked_java(&services.db, &executable, Some(major)).await;
}

3
packages/oneclient_core/src/java/vendors/mod.rs vendored
View file

@ -50,8 +50,9 @@ pub trait JavaRuntimeProvider: Send + Sync {
&self,
package: &JavaPackage,
services: &LauncherServices,
progress: Option<&crate::notification::GroupedProgressSession>,
) -> LauncherResult<PathBuf> {
super::install_package(package, services).await
super::install_package(package, services, progress).await
}
}

2
packages/oneclient_core/src/notification/data.rs
View file

@ -74,6 +74,7 @@ pub enum TaskCategory {
Libraries,
Natives,
Assets,
Java,
#[default]
Packages,
}
@ -86,6 +87,7 @@ impl TaskCategory {
Self::Libraries => "Libraries",
Self::Natives => "Natives",
Self::Assets => "Assets",
Self::Java => "Java runtime",
Self::Packages => "Packages",
}
}

100
packages/oneclient_core/src/notification/grouped.rs
View file

@ -218,7 +218,105 @@ impl GroupedProgressChild {
}
impl Drop for GroupedProgressChild {
/// Only the *last* handle finishes the task.
///
/// `finished` lives in the shared inner, so finishing on every clone's drop
/// ended the task the moment any temporary copy died — and copies are handed
/// around routinely (into `ResponseOptions`, into stream closures). A
/// download would emit `FinishChild` before its first byte arrived, after
/// which every `set_progress` silently no-opped and the UI removed the row.
/// The safety net for early returns still works: on an error path the last
/// handle goes out of scope too.
fn drop(&mut self) {
self.finish();
if Arc::strong_count(&self.inner) == 1 {
self.finish();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::notification::Notification;
fn drain(
rx: &mut tokio::sync::mpsc::UnboundedReceiver<Notification>,
) -> Vec<GroupedProgressEvent> {
let mut events = Vec::new();
while let Ok(notification) = rx.try_recv() {
if let Notification::GroupedProgress(event) = notification {
events.push(event);
}
}
events
}
fn has_finish(events: &[GroupedProgressEvent]) -> bool {
events
.iter()
.any(|e| matches!(e, GroupedProgressEvent::FinishChild { .. }))
}
/// A child handle gets cloned into `ResponseOptions` and into the streaming
/// closure. If dropping one of those copies finished the task, the download
/// would report completion before its first byte and every later progress
/// update would be dropped on the floor.
#[test]
fn dropping_a_clone_does_not_finish_the_task() {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let notifier = NotificationService::new(tx);
let session = GroupedProgressSession::start(&notifier, "test");
let child = session.child("file", 100, TaskCategory::Assets);
let _ = drain(&mut rx);
drop(child.clone());
assert!(
!has_finish(&drain(&mut rx)),
"a cloned handle going out of scope must not finish the task"
);
child.set_progress(50, Some(100));
let events = drain(&mut rx);
assert!(
events.iter().any(|e| matches!(
e,
GroupedProgressEvent::UpdateChild { current: 50, .. }
)),
"progress must still be reported after a clone is dropped: {events:?}"
);
drop(child);
assert!(
has_finish(&drain(&mut rx)),
"the last handle going out of scope should finish the task"
);
}
#[test]
fn explicit_finish_reports_once() {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let notifier = NotificationService::new(tx);
let session = GroupedProgressSession::start(&notifier, "test");
let child = session.child("file", 100, TaskCategory::Assets);
let _ = drain(&mut rx);
child.finish();
assert_eq!(drain(&mut rx).iter().filter(|e| matches!(e, GroupedProgressEvent::FinishChild { .. })).count(), 1);
drop(child);
assert!(!has_finish(&drain(&mut rx)), "drop must not finish twice");
}
#[test]
fn progress_after_finish_is_ignored() {
let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
let notifier = NotificationService::new(tx);
let session = GroupedProgressSession::start(&notifier, "test");
let child = session.child("file", 100, TaskCategory::Assets);
child.finish();
let _ = drain(&mut rx);
child.set_progress(75, Some(100));
assert!(drain(&mut rx).is_empty());
}
}

1
packages/oneclient_core/src/paths.rs
View file

@ -7,6 +7,7 @@ use crate::LauncherResult;
const QUALIFIER: &str = "org";
const ORGANIZATION: &str = "Polyfrost";
#[cfg(not(debug_assertions))]
const APPLICATION: &str = "OneClient";
// Dev builds use a separate launcher dir (e.g. `oneclient-dev`) so a running

150
packages/oneclient_core/tests/download_verify.rs Normal file
View file

@ -0,0 +1,150 @@
use oneclient_core::crypto::{Sha1Stream, sha1_bytes, sha1_file};
use oneclient_core::dev;
use oneclient_core::game::download_to_path;
use oneclient_core::notification::{GroupedProgressSession, TaskCategory};
/// A Minecraft asset object. Objects are content-addressed and immutable, so
/// this URL/hash pair stays valid.
const ASSET_SHA1: &str = "af96f55a90eaf11b327f1b5f8834a051027dc506";
const ASSET_URL: &str =
"https://resources.download.minecraft.net/af/af96f55a90eaf11b327f1b5f8834a051027dc506";
const ASSET_SIZE: u64 = 2063;
/// Scratch directory unique to one test, removed by [`Scratch`]'s drop.
struct Scratch(std::path::PathBuf);
impl Scratch {
fn new(tag: &str) -> Self {
static COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
let n = COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
let path =
std::env::temp_dir().join(format!("oneclient-{tag}-{}-{n}", std::process::id()));
std::fs::create_dir_all(&path).unwrap();
Self(path)
}
fn join(&self, name: &str) -> std::path::PathBuf {
self.0.join(name)
}
}
impl Drop for Scratch {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.0);
}
}
#[test]
fn sha1_stream_matches_one_shot_hash() {
let data: Vec<u8> = (0..70_000u32).map(|i| (i % 251) as u8).collect();
let mut hasher = Sha1Stream::new();
for chunk in data.chunks(4096) {
hasher.update(chunk);
}
assert_eq!(hasher.finish(), sha1_bytes(&data));
}
#[test]
fn sha1_stream_of_empty_input_matches() {
assert_eq!(Sha1Stream::new().finish(), sha1_bytes(&[]));
}
#[tokio::test]
async fn sha1_file_matches_sha1_bytes() {
let dir = Scratch::new("sha1-file");
let path = dir.join("payload.bin");
// Larger than the hashing buffer, so the multi-read path is covered too.
let data: Vec<u8> = (0..600_000u32).map(|i| (i % 253) as u8).collect();
std::fs::write(&path, &data).unwrap();
assert_eq!(sha1_file(&path).await.unwrap(), sha1_bytes(&data));
}
#[tokio::test]
async fn sha1_file_handles_empty_file() {
let dir = Scratch::new("sha1-empty");
let path = dir.join("empty.bin");
std::fs::write(&path, []).unwrap();
assert_eq!(sha1_file(&path).await.unwrap(), sha1_bytes(&[]));
}
#[tokio::test]
#[ignore = "requires network"]
async fn download_to_path_accepts_matching_hash() {
let services = dev::ephemeral_services().await.unwrap();
let progress = GroupedProgressSession::start(&services.notifier, "test");
let dir = Scratch::new("download-ok");
let dest = dir.join("nested").join("icon.png");
download_to_path(
&services.requester,
&services.notifier,
&progress,
"icon",
TaskCategory::Assets,
ASSET_SIZE,
ASSET_URL,
&dest,
Some(ASSET_SHA1),
)
.await
.expect("download should verify");
assert_eq!(std::fs::metadata(&dest).unwrap().len(), ASSET_SIZE);
assert_eq!(sha1_file(&dest).await.unwrap(), ASSET_SHA1);
}
#[tokio::test]
#[ignore = "requires network"]
async fn download_to_path_rejects_mismatched_hash_and_removes_file() {
let services = dev::ephemeral_services().await.unwrap();
let progress = GroupedProgressSession::start(&services.notifier, "test");
let dir = Scratch::new("download-bad");
let dest = dir.join("icon.png");
let err = download_to_path(
&services.requester,
&services.notifier,
&progress,
"icon",
TaskCategory::Assets,
ASSET_SIZE,
ASSET_URL,
&dest,
Some("0000000000000000000000000000000000000000"),
)
.await
.expect_err("mismatched hash should fail");
assert!(err.to_string().contains("0000000000000000"), "{err}");
assert!(!dest.exists(), "corrupt download should be removed");
}
#[tokio::test]
#[ignore = "requires network"]
async fn download_to_path_without_expected_hash_still_writes() {
let services = dev::ephemeral_services().await.unwrap();
let progress = GroupedProgressSession::start(&services.notifier, "test");
let dir = Scratch::new("download-nohash");
let dest = dir.join("icon.png");
download_to_path(
&services.requester,
&services.notifier,
&progress,
"icon",
TaskCategory::Assets,
0,
ASSET_URL,
&dest,
None,
)
.await
.expect("download should succeed");
assert_eq!(sha1_file(&dest).await.unwrap(), ASSET_SHA1);
}

52
packages/polyio/src/file.rs
View file

@ -5,6 +5,29 @@ use serde::{Serialize, de::DeserializeOwned};
use crate::{IOError, PolyIOResult};
/// Smallest write buffer. HTTP chunks already arrive at ~16 KiB, so anything
/// below this is buffering that was already the right size.
const MIN_WRITE_BUFFER: usize = 16 * 1024;
/// Largest write buffer, for multi-megabyte downloads.
const MAX_WRITE_BUFFER: usize = 256 * 1024;
/// Used when the response has no length: assume something mid-sized.
const DEFAULT_WRITE_BUFFER: usize = 64 * 1024;
/// Picks a write buffer that fits the file rather than a fixed large one.
///
/// This is a memory-footprint choice, not a throughput one: measured against
/// the real asset-size distribution, 64 KiB / 256 KiB / file-sized all write at
/// the same rate (~200 MB/s, ~25x faster than the download feeding them), so
/// the buffer is nowhere near the critical path. What a fixed 256 KiB *does*
/// cost is 8 MiB of buffers across 32 concurrent asset downloads, each holding
/// a ~10 KiB object.
fn write_buffer_size(size_hint: Option<u64>) -> usize {
match size_hint {
Some(0) | None => DEFAULT_WRITE_BUFFER,
Some(size) => (size.min(MAX_WRITE_BUFFER as u64) as usize).max(MIN_WRITE_BUFFER),
}
}
/// Returns a stream over the entries within a directory.
#[tracing::instrument(
level = "debug",
@ -212,7 +235,8 @@ where
)]
pub async fn write_stream<S, E>(
path: impl AsRef<std::path::Path>,
mut stream: S
mut stream: S,
size_hint: Option<u64>,
) -> Result<(), E>
where
S: futures_lite::Stream<Item = Result<bytes::Bytes, E>> + Unpin + Send,
@ -221,7 +245,7 @@ where
let path = path.as_ref();
let file = tokio::fs::File::create(path).await.map_err(IOError::from)?;
let mut writer = tokio::io::BufWriter::new(file);
let mut writer = tokio::io::BufWriter::with_capacity(write_buffer_size(size_hint), file);
let mut write_result = Ok(());
while let Some(chunk_result) = futures_lite::StreamExt::next(&mut stream).await {
@ -472,4 +496,26 @@ pub fn sanitize_path(path: impl AsRef<std::path::Path>) -> PathBuf {
)]
pub async fn stat(path: impl AsRef<std::path::Path>) -> PolyIOResult<Metadata> {
tokio::fs::metadata(path).await.map_err(IOError::from)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn write_buffer_fits_the_file() {
// A typical Minecraft asset object.
assert_eq!(write_buffer_size(Some(10_217)), MIN_WRITE_BUFFER);
assert_eq!(write_buffer_size(Some(1)), MIN_WRITE_BUFFER);
// Mid-sized files get exactly what they need.
assert_eq!(write_buffer_size(Some(64 * 1024)), 64 * 1024);
// Large downloads are capped rather than matched.
assert_eq!(write_buffer_size(Some(25_000_000)), MAX_WRITE_BUFFER);
assert_eq!(write_buffer_size(Some(u64::MAX)), MAX_WRITE_BUFFER);
}
#[test]
fn write_buffer_falls_back_without_a_length() {
assert_eq!(write_buffer_size(None), DEFAULT_WRITE_BUFFER);
assert_eq!(write_buffer_size(Some(0)), DEFAULT_WRITE_BUFFER);
}
}