Ryan/drop-app
1
0
Fork 0
mirror of https://github.com/Drop-OSS/drop-app.git synced 2025-11-16 01:31:22 +10:00
Code Issues Packages Projects Releases Wiki Activity

v2 download API and fixes (#112)

Browse Source 
* fix: potential download fixes

* fix: show installed games not on remote

* fix: more download_logic error handling

* partial: move to async

* feat: interactivity improvements

* feat: v2 download API

* fix: download seek offsets

* fix: clippy

* fix: apply clippy suggestion

* fix: performance improvements starting up download

* fix: finished bucket file

* fix: ui tweaks and fixes

* fix: revert version to 0.3.2

* fix: clippy
This commit is contained in:
DecDuck
2025-08-09 15:50:21 +10:00
committed by GitHub
parent 3b830e2a44
commit 16365713cf
27 changed files with 859 additions and 603 deletions
Download Patch File Download Diff File Expand all files Collapse all files

69
src-tauri/src/games/downloads/commands.rs
View File

@ -3,44 +3,47 @@ use std::{
sync::{Arc, Mutex},
};
use crate::{
database::{db::borrow_db_checked, models::data::GameDownloadStatus},
download_manager::
downloadable::Downloadable
,
error::application_download_error::ApplicationDownloadError,
AppState,
database::{
db::borrow_db_checked,
models::data::GameDownloadStatus,
},
download_manager::downloadable::Downloadable,
error::application_download_error::ApplicationDownloadError,
};
use super::download_agent::GameDownloadAgent;
#[tauri::command]
pub fn download_game(
pub async fn download_game(
game_id: String,
game_version: String,
install_dir: usize,
state: tauri::State<'_, Mutex<AppState>>,
state: tauri::State<'_, Mutex<AppState<'_>>>,
) -> Result<(), ApplicationDownloadError> {
let sender = state.lock().unwrap().download_manager.get_sender();
let game_download_agent = GameDownloadAgent::new_from_index(
game_id,
game_version,
install_dir,
sender,
)?;
let game_download_agent = Arc::new(Box::new(game_download_agent) as Box<dyn Downloadable + Send + Sync>);
let sender = { state.lock().unwrap().download_manager.get_sender().clone() };
let game_download_agent =
GameDownloadAgent::new_from_index(game_id.clone(), game_version.clone(), install_dir, sender).await?;
let game_download_agent =
Arc::new(Box::new(game_download_agent) as Box<dyn Downloadable + Send + Sync>);
state
.lock()
.unwrap()
.download_manager
.queue_download(game_download_agent).unwrap();
.lock()
.unwrap()
.download_manager
.queue_download(game_download_agent.clone())
.unwrap();
Ok(())
}
#[tauri::command]
pub fn resume_download(
pub async fn resume_download(
game_id: String,
state: tauri::State<'_, Mutex<AppState>>,
state: tauri::State<'_, Mutex<AppState<'_>>>,
) -> Result<(), ApplicationDownloadError> {
let s = borrow_db_checked()
.applications
@ -62,17 +65,21 @@ pub fn resume_download(
let sender = state.lock().unwrap().download_manager.get_sender();
let parent_dir: PathBuf = install_dir.into();
let game_download_agent = Arc::new(Box::new(GameDownloadAgent::new(
game_id,
version_name.clone(),
parent_dir.parent().unwrap().to_path_buf(),
sender,
)?) as Box<dyn Downloadable + Send + Sync>);
let game_download_agent = Arc::new(Box::new(
GameDownloadAgent::new(
game_id,
version_name.clone(),
parent_dir.parent().unwrap().to_path_buf(),
sender,
)
.await?,
) as Box<dyn Downloadable + Send + Sync>);
state
.lock()
.unwrap()
.download_manager
.queue_download(game_download_agent).unwrap();
.lock()
.unwrap()
.download_manager
.queue_download(game_download_agent)
.unwrap();
Ok(())
}

333
src-tauri/src/games/downloads/download_agent.rs
View File

@ -11,13 +11,15 @@ use crate::download_manager::util::download_thread_control_flag::{
use crate::download_manager::util::progress_object::{ProgressHandle, ProgressObject};
use crate::error::application_download_error::ApplicationDownloadError;
use crate::error::remote_access_error::RemoteAccessError;
use crate::games::downloads::manifest::{DropDownloadContext, DropManifest};
use crate::games::downloads::manifest::{
DownloadBucket, DownloadContext, DownloadDrop, DropManifest, DropValidateContext, ManifestBody,
};
use crate::games::downloads::validate::validate_game_chunk;
use crate::games::library::{on_game_complete, push_game_update, set_partially_installed};
use crate::games::state::GameStatusManager;
use crate::process::utils::get_disk_available;
use crate::remote::requests::make_request;
use crate::remote::utils::DROP_CLIENT_SYNC;
use crate::remote::requests::generate_url;
use crate::remote::utils::{DROP_CLIENT_ASYNC, DROP_CLIENT_SYNC};
use log::{debug, error, info, warn};
use rayon::ThreadPoolBuilder;
use std::collections::HashMap;
@ -31,16 +33,18 @@ use tauri::{AppHandle, Emitter};
#[cfg(target_os = "linux")]
use rustix::fs::{FallocateFlags, fallocate};
use super::download_logic::download_game_chunk;
use super::download_logic::download_game_bucket;
use super::drop_data::DropData;
static RETRY_COUNT: usize = 3;
const TARGET_BUCKET_SIZE: usize = 63 * 1000 * 1000;
pub struct GameDownloadAgent {
pub id: String,
pub version: String,
pub control_flag: DownloadThreadControl,
contexts: Mutex<Vec<DropDownloadContext>>,
buckets: Mutex<Vec<DownloadBucket>>,
context_map: Mutex<HashMap<String, bool>>,
pub manifest: Mutex<Option<DropManifest>>,
pub progress: Arc<ProgressObject>,
@ -50,19 +54,21 @@ pub struct GameDownloadAgent {
}
impl GameDownloadAgent {
pub fn new_from_index(
pub async fn new_from_index(
id: String,
version: String,
target_download_dir: usize,
sender: Sender<DownloadManagerSignal>,
) -> Result<Self, ApplicationDownloadError> {
let db_lock = borrow_db_checked();
let base_dir = db_lock.applications.install_dirs[target_download_dir].clone();
drop(db_lock);
let base_dir = {
let db_lock = borrow_db_checked();
Self::new(id, version, base_dir, sender)
db_lock.applications.install_dirs[target_download_dir].clone()
};
Self::new(id, version, base_dir, sender).await
}
pub fn new(
pub async fn new(
id: String,
version: String,
base_dir: PathBuf,
@ -82,7 +88,7 @@ impl GameDownloadAgent {
version,
control_flag,
manifest: Mutex::new(None),
contexts: Mutex::new(Vec::new()),
buckets: Mutex::new(Vec::new()),
context_map: Mutex::new(HashMap::new()),
progress: Arc::new(ProgressObject::new(0, 0, sender.clone())),
sender,
@ -90,7 +96,7 @@ impl GameDownloadAgent {
status: Mutex::new(DownloadStatus::Queued),
};
result.ensure_manifest_exists()?;
result.ensure_manifest_exists().await?;
let required_space = result
.manifest
@ -100,8 +106,7 @@ impl GameDownloadAgent {
.unwrap()
.values()
.map(|e| e.lengths.iter().sum::<usize>())
.sum::<usize>()
as u64;
.sum::<usize>() as u64;
let available_space = get_disk_available(data_base_dir_path)? as u64;
@ -117,26 +122,25 @@ impl GameDownloadAgent {
// Blocking
pub fn setup_download(&self, app_handle: &AppHandle) -> Result<(), ApplicationDownloadError> {
self.ensure_manifest_exists()?;
let mut db_lock = borrow_db_mut_checked();
let status = ApplicationTransientStatus::Downloading {
version_name: self.version.clone(),
};
db_lock
.applications
.transient_statuses
.insert(self.metadata(), status.clone());
// Don't use GameStatusManager because this game isn't installed
push_game_update(app_handle, &self.metadata().id, None, (None, Some(status)));
self.ensure_contexts()?;
if !self.check_manifest_exists() {
return Err(ApplicationDownloadError::NotInitialized);
}
self.ensure_buckets()?;
self.control_flag.set(DownloadThreadControlFlag::Go);
let mut db_lock = borrow_db_mut_checked();
db_lock.applications.transient_statuses.insert(
self.metadata(),
ApplicationTransientStatus::Downloading {
version_name: self.version.clone(),
},
);
push_game_update(
app_handle,
&self.metadata().id,
None,
GameStatusManager::fetch_state(&self.metadata().id, &db_lock),
);
Ok(())
}
@ -147,9 +151,7 @@ impl GameDownloadAgent {
info!("beginning download for {}...", self.metadata().id);
let res = self
.run()
.map_err(|()| ApplicationDownloadError::DownloadError);
let res = self.run().map_err(ApplicationDownloadError::Communication);
debug!(
"{} took {}ms to download",
@ -159,37 +161,43 @@ impl GameDownloadAgent {
res
}
pub fn ensure_manifest_exists(&self) -> Result<(), ApplicationDownloadError> {
pub fn check_manifest_exists(&self) -> bool {
self.manifest.lock().unwrap().is_some()
}
pub async fn ensure_manifest_exists(&self) -> Result<(), ApplicationDownloadError> {
if self.manifest.lock().unwrap().is_some() {
return Ok(());
}
self.download_manifest()
self.download_manifest().await
}
fn download_manifest(&self) -> Result<(), ApplicationDownloadError> {
let header = generate_authorization_header();
let client = DROP_CLIENT_SYNC.clone();
let response = make_request(
&client,
async fn download_manifest(&self) -> Result<(), ApplicationDownloadError> {
let client = DROP_CLIENT_ASYNC.clone();
let url = generate_url(
&["/api/v1/client/game/manifest"],
&[("id", &self.id), ("version", &self.version)],
|f| f.header("Authorization", header),
)
.map_err(ApplicationDownloadError::Communication)?
.send()
.map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
.map_err(ApplicationDownloadError::Communication)?;
let response = client
.get(url)
.header("Authorization", generate_authorization_header())
.send()
.await
.map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
if response.status() != 200 {
return Err(ApplicationDownloadError::Communication(
RemoteAccessError::ManifestDownloadFailed(
response.status(),
response.text().unwrap(),
response.text().await.unwrap(),
),
));
}
let manifest_download: DropManifest = response.json().unwrap();
let manifest_download: DropManifest = response.json().await.unwrap();
if let Ok(mut manifest) = self.manifest.lock() {
*manifest = Some(manifest_download);
@ -201,20 +209,23 @@ impl GameDownloadAgent {
// Sets it up for both download and validate
fn setup_progress(&self) {
let contexts = self.contexts.lock().unwrap();
let buckets = self.buckets.lock().unwrap();
let length = contexts.len();
let chunk_count = buckets.iter().map(|e| e.drops.len()).sum();
let chunk_count = contexts.iter().map(|chunk| chunk.length).sum();
let total_length = buckets
.iter()
.map(|bucket| bucket.drops.iter().map(|e| e.length).sum::<usize>())
.sum();
self.progress.set_max(chunk_count);
self.progress.set_size(length);
self.progress.set_max(total_length);
self.progress.set_size(chunk_count);
self.progress.reset();
}
pub fn ensure_contexts(&self) -> Result<(), ApplicationDownloadError> {
if self.contexts.lock().unwrap().is_empty() {
self.generate_contexts()?;
pub fn ensure_buckets(&self) -> Result<(), ApplicationDownloadError> {
if self.buckets.lock().unwrap().is_empty() {
self.generate_buckets()?;
}
*self.context_map.lock().unwrap() = self.dropdata.get_contexts();
@ -222,14 +233,22 @@ impl GameDownloadAgent {
Ok(())
}
pub fn generate_contexts(&self) -> Result<(), ApplicationDownloadError> {
pub fn generate_buckets(&self) -> Result<(), ApplicationDownloadError> {
let manifest = self.manifest.lock().unwrap().clone().unwrap();
let game_id = self.id.clone();
let mut contexts = Vec::new();
let base_path = Path::new(&self.dropdata.base_path);
create_dir_all(base_path).unwrap();
let mut buckets = Vec::new();
let mut current_bucket = DownloadBucket {
game_id: game_id.clone(),
version: self.version.clone(),
drops: Vec::new(),
};
let mut current_bucket_size = 0;
for (raw_path, chunk) in manifest {
let path = base_path.join(Path::new(&raw_path));
@ -244,42 +263,79 @@ impl GameDownloadAgent {
.truncate(false)
.open(path.clone())
.unwrap();
let mut running_offset = 0;
let mut file_running_offset = 0;
for (index, length) in chunk.lengths.iter().enumerate() {
contexts.push(DropDownloadContext {
file_name: raw_path.to_string(),
version: chunk.version_name.to_string(),
offset: running_offset,
index,
game_id: game_id.to_string(),
path: path.clone(),
checksum: chunk.checksums[index].clone(),
let drop = DownloadDrop {
filename: raw_path.to_string(),
start: file_running_offset,
length: *length,
checksum: chunk.checksums[index].clone(),
permissions: chunk.permissions,
});
running_offset += *length as u64;
path: path.clone(),
index,
};
file_running_offset += *length;
if *length >= TARGET_BUCKET_SIZE {
// They get their own bucket
buckets.push(DownloadBucket {
game_id: game_id.clone(),
version: self.version.clone(),
drops: vec![drop],
});
continue;
}
if current_bucket_size + *length >= TARGET_BUCKET_SIZE
&& !current_bucket.drops.is_empty()
{
// Move current bucket into list and make a new one
buckets.push(current_bucket);
current_bucket = DownloadBucket {
game_id: game_id.clone(),
version: self.version.clone(),
drops: Vec::new(),
};
current_bucket_size = 0;
}
current_bucket.drops.push(drop);
current_bucket_size += *length;
}
#[cfg(target_os = "linux")]
if running_offset > 0 && !already_exists {
let _ = fallocate(file, FallocateFlags::empty(), 0, running_offset);
if file_running_offset > 0 && !already_exists {
let _ = fallocate(file, FallocateFlags::empty(), 0, file_running_offset as u64);
}
}
let existing_contexts = self.dropdata.get_completed_contexts();
if !current_bucket.drops.is_empty() {
buckets.push(current_bucket);
}
info!("buckets: {}", buckets.len());
let existing_contexts = self.dropdata.get_contexts();
self.dropdata.set_contexts(
&contexts
&buckets
.iter()
.map(|x| (x.checksum.clone(), existing_contexts.contains(&x.checksum)))
.flat_map(|x| x.drops.iter().map(|v| v.checksum.clone()))
.map(|x| {
let contains = existing_contexts.get(&x).unwrap_or(&false);
(x, *contains)
})
.collect::<Vec<(String, bool)>>(),
);
*self.contexts.lock().unwrap() = contexts;
*self.buckets.lock().unwrap() = buckets;
Ok(())
}
fn run(&self) -> Result<bool, ()> {
fn run(&self) -> Result<bool, RemoteAccessError> {
self.setup_progress();
let max_download_threads = borrow_db_checked().settings.max_download_threads;
@ -295,78 +351,81 @@ impl GameDownloadAgent {
let completed_contexts = Arc::new(boxcar::Vec::new());
let completed_indexes_loop_arc = completed_contexts.clone();
let contexts = self.contexts.lock().unwrap();
let download_context = DROP_CLIENT_SYNC
.post(generate_url(&["/api/v2/client/context"], &[]).unwrap())
.json(&ManifestBody {
game: self.id.clone(),
version: self.version.clone(),
})
.header("Authorization", generate_authorization_header())
.send()?;
if download_context.status() != 200 {
return Err(RemoteAccessError::InvalidResponse(download_context.json()?));
}
let download_context = &download_context.json::<DownloadContext>()?;
info!("download context: {}", download_context.context);
let buckets = self.buckets.lock().unwrap();
pool.scope(|scope| {
let client = &DROP_CLIENT_SYNC.clone();
let context_map = self.context_map.lock().unwrap();
for (index, context) in contexts.iter().enumerate() {
let client = client.clone();
let completed_indexes = completed_indexes_loop_arc.clone();
for (index, bucket) in buckets.iter().enumerate() {
let mut bucket = (*bucket).clone();
let completed_contexts = completed_indexes_loop_arc.clone();
let progress = self.progress.get(index);
let progress_handle = ProgressHandle::new(progress, self.progress.clone());
// If we've done this one already, skip it
// Note to future DecDuck, DropData gets loaded into context_map
if let Some(v) = context_map.get(&context.checksum)
&& *v
{
progress_handle.skip(context.length);
let todo_drops = bucket
.drops
.into_iter()
.filter(|e| {
let todo = !*context_map.get(&e.checksum).unwrap_or(&false);
if !todo {
progress_handle.skip(e.length);
}
todo
})
.collect::<Vec<DownloadDrop>>();
if todo_drops.is_empty() {
continue;
}
};
bucket.drops = todo_drops;
let sender = self.sender.clone();
let request = match make_request(
&client,
&["/api/v1/client/chunk"],
&[
("id", &context.game_id),
("version", &context.version),
("name", &context.file_name),
("chunk", &context.index.to_string()),
],
|r| r,
) {
Ok(request) => request,
Err(e) => {
sender
.send(DownloadManagerSignal::Error(
ApplicationDownloadError::Communication(e),
))
.unwrap();
continue;
}
};
scope.spawn(move |_| {
// 3 attempts
for i in 0..RETRY_COUNT {
let loop_progress_handle = progress_handle.clone();
match download_game_chunk(
context,
match download_game_bucket(
&bucket,
download_context,
&self.control_flag,
loop_progress_handle,
request.try_clone().unwrap(),
) {
Ok(true) => {
completed_indexes.push(context.checksum.clone());
for drop in bucket.drops {
completed_contexts.push(drop.checksum);
}
return;
}
Ok(false) => return,
Err(e) => {
warn!("game download agent error: {e}");
let retry = match &e {
ApplicationDownloadError::Communication(
_remote_access_error,
) => true,
ApplicationDownloadError::Checksum => true,
ApplicationDownloadError::Lock => true,
ApplicationDownloadError::IoError(_error_kind) => false,
ApplicationDownloadError::DownloadError => false,
ApplicationDownloadError::DiskFull(_, _) => false,
};
let retry = matches!(
&e,
ApplicationDownloadError::Communication(_)
| ApplicationDownloadError::Checksum
| ApplicationDownloadError::Lock
);
if i == RETRY_COUNT - 1 || !retry {
warn!("retry logic failed, not re-attempting.");
@ -390,14 +449,14 @@ impl GameDownloadAgent {
context_map_lock.values().filter(|x| **x).count()
};
let context_map_lock = self.context_map.lock().unwrap();
let contexts = contexts
let contexts = buckets
.iter()
.flat_map(|x| x.drops.iter().map(|e| e.checksum.clone()))
.map(|x| {
(
x.checksum.clone(),
context_map_lock.get(&x.checksum).copied().unwrap_or(false),
)
let completed = context_map_lock.get(&x).unwrap_or(&false);
(x, *completed)
})
.collect::<Vec<(String, bool)>>();
drop(context_map_lock);
@ -408,10 +467,11 @@ impl GameDownloadAgent {
// If there are any contexts left which are false
if !contexts.iter().all(|x| x.1) {
info!(
"download agent for {} exited without completing ({}/{})",
"download agent for {} exited without completing ({}/{}) ({} buckets)",
self.id.clone(),
completed_lock_len,
contexts.len(),
buckets.len()
);
return Ok(false);
}
@ -442,13 +502,15 @@ impl GameDownloadAgent {
pub fn validate(&self, app_handle: &AppHandle) -> Result<bool, ApplicationDownloadError> {
self.setup_validate(app_handle);
let contexts = self.contexts.lock().unwrap();
let buckets = self.buckets.lock().unwrap();
let contexts: Vec<DropValidateContext> = buckets
.clone()
.into_iter()
.flat_map(|e| -> Vec<DropValidateContext> { e.into() })
.collect();
let max_download_threads = borrow_db_checked().settings.max_download_threads;
debug!(
"validating game: {} with {} threads",
self.dropdata.game_id, max_download_threads
);
info!("{} validation contexts", contexts.len());
let pool = ThreadPoolBuilder::new()
.num_threads(max_download_threads)
.build()
@ -549,6 +611,13 @@ impl Downloadable for GameDownloadAgent {
.applications
.transient_statuses
.remove(&self.metadata());
push_game_update(
app_handle,
&self.id,
None,
GameStatusManager::fetch_state(&self.id, &handle),
);
}
fn on_complete(&self, app_handle: &tauri::AppHandle) {

210
src-tauri/src/games/downloads/download_logic.rs
View File

@ -5,13 +5,15 @@ use crate::download_manager::util::progress_object::ProgressHandle;
use crate::error::application_download_error::ApplicationDownloadError;
use crate::error::drop_server_error::DropServerError;
use crate::error::remote_access_error::RemoteAccessError;
use crate::games::downloads::manifest::DropDownloadContext;
use crate::games::downloads::manifest::{ChunkBody, DownloadBucket, DownloadContext, DownloadDrop};
use crate::remote::auth::generate_authorization_header;
use log::{debug, warn};
use crate::remote::requests::generate_url;
use crate::remote::utils::DROP_CLIENT_SYNC;
use log::{info, warn};
use md5::{Context, Digest};
use reqwest::blocking::{RequestBuilder, Response};
use reqwest::blocking::Response;
use std::fs::{set_permissions, Permissions};
use std::fs::{Permissions, set_permissions};
use std::io::Read;
#[cfg(unix)]
use std::os::unix::fs::PermissionsExt;
@ -21,21 +23,29 @@ use std::{
path::PathBuf,
};
static MAX_PACKET_LENGTH: usize = 4096 * 4;
pub struct DropWriter<W: Write> {
hasher: Context,
destination: W,
destination: BufWriter<W>,
progress: ProgressHandle,
}
impl DropWriter<File> {
fn new(path: PathBuf) -> Self {
let destination = OpenOptions::new().write(true).create(true).truncate(false).open(&path).unwrap();
Self {
destination,
fn new(path: PathBuf, progress: ProgressHandle) -> Result<Self, io::Error> {
let destination = OpenOptions::new()
.write(true)
.create(true)
.truncate(false)
.open(&path)?;
Ok(Self {
destination: BufWriter::with_capacity(1024 * 1024, destination),
hasher: Context::new(),
}
progress,
})
}
fn finish(mut self) -> io::Result<Digest> {
self.flush().unwrap();
self.flush()?;
Ok(self.hasher.compute())
}
}
@ -45,7 +55,10 @@ impl Write for DropWriter<File> {
self.hasher
.write_all(buf)
.map_err(|e| io::Error::other(format!("Unable to write to hasher: {e}")))?;
self.destination.write(buf)
let bytes_written = self.destination.write(buf)?;
self.progress.add(bytes_written);
Ok(bytes_written)
}
fn flush(&mut self) -> io::Result<()> {
@ -62,91 +75,103 @@ impl Seek for DropWriter<File> {
pub struct DropDownloadPipeline<'a, R: Read, W: Write> {
pub source: R,
pub destination: DropWriter<W>,
pub drops: Vec<DownloadDrop>,
pub destination: Vec<DropWriter<W>>,
pub control_flag: &'a DownloadThreadControl,
pub progress: ProgressHandle,
pub size: usize,
}
impl<'a> DropDownloadPipeline<'a, Response, File> {
fn new(
source: Response,
destination: DropWriter<File>,
drops: Vec<DownloadDrop>,
control_flag: &'a DownloadThreadControl,
progress: ProgressHandle,
size: usize,
) -> Self {
Self {
) -> Result<Self, io::Error> {
Ok(Self {
source,
destination,
destination: drops
.iter()
.map(|drop| DropWriter::new(drop.path.clone(), progress.clone()))
.try_collect()?,
drops,
control_flag,
progress,
size,
}
})
}
fn copy(&mut self) -> Result<bool, io::Error> {
let copy_buf_size = 512;
let mut copy_buf = vec![0; copy_buf_size];
let mut buf_writer = BufWriter::with_capacity(1024 * 1024, &mut self.destination);
let mut copy_buffer = [0u8; MAX_PACKET_LENGTH];
for (index, drop) in self.drops.iter().enumerate() {
let destination = self
.destination
.get_mut(index)
.ok_or(io::Error::other("no destination"))
.unwrap();
let mut remaining = drop.length;
if drop.start != 0 {
destination.seek(SeekFrom::Start(drop.start.try_into().unwrap()))?;
}
loop {
let size = MAX_PACKET_LENGTH.min(remaining);
self.source.read_exact(&mut copy_buffer[0..size])?;
remaining -= size;
destination.write_all(&copy_buffer[0..size])?;
if remaining == 0 {
break;
};
}
let mut current_size = 0;
loop {
if self.control_flag.get() == DownloadThreadControlFlag::Stop {
buf_writer.flush()?;
return Ok(false);
}
let mut bytes_read = self.source.read(&mut copy_buf)?;
current_size += bytes_read;
if current_size > self.size {
let over = current_size - self.size;
warn!("server sent too many bytes... {over} over");
bytes_read -= over;
current_size = self.size;
}
buf_writer.write_all(&copy_buf[0..bytes_read])?;
self.progress.add(bytes_read);
if current_size >= self.size {
debug!(
"finished with final size of {} vs {}",
current_size, self.size
);
break;
}
}
buf_writer.flush()?;
Ok(true)
}
fn finish(self) -> Result<Digest, io::Error> {
let checksum = self.destination.finish()?;
Ok(checksum)
fn finish(self) -> Result<Vec<Digest>, io::Error> {
let checksums = self
.destination
.into_iter()
.map(|e| e.finish())
.try_collect()?;
Ok(checksums)
}
}
pub fn download_game_chunk(
ctx: &DropDownloadContext,
pub fn download_game_bucket(
bucket: &DownloadBucket,
ctx: &DownloadContext,
control_flag: &DownloadThreadControl,
progress: ProgressHandle,
request: RequestBuilder,
) -> Result<bool, ApplicationDownloadError> {
// If we're paused
if control_flag.get() == DownloadThreadControlFlag::Stop {
progress.set(0);
return Ok(false);
}
let response = request
.header("Authorization", generate_authorization_header())
let header = generate_authorization_header();
let url = generate_url(&["/api/v2/client/chunk"], &[])
.map_err(ApplicationDownloadError::Communication)?;
let body = ChunkBody::create(ctx, &bucket.drops);
let response = DROP_CLIENT_SYNC
.post(url)
.json(&body)
.header("Authorization", header)
.send()
.map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
if response.status() != 200 {
debug!("chunk request got status code: {}", response.status());
let raw_res = response.text().unwrap();
info!("chunk request got status code: {}", response.status());
let raw_res = response.text().map_err(|e| {
ApplicationDownloadError::Communication(RemoteAccessError::FetchError(e.into()))
})?;
info!("{}", raw_res);
if let Ok(err) = serde_json::from_str::<DropServerError>(&raw_res) {
return Err(ApplicationDownloadError::Communication(
RemoteAccessError::InvalidResponse(err),
@ -157,30 +182,35 @@ pub fn download_game_chunk(
));
}
let mut destination = DropWriter::new(ctx.path.clone());
let lengths = response
.headers()
.get("Content-Lengths")
.ok_or(ApplicationDownloadError::Communication(
RemoteAccessError::UnparseableResponse("missing Content-Lengths header".to_owned()),
))?
.to_str()
.unwrap();
if ctx.offset != 0 {
destination
.seek(SeekFrom::Start(ctx.offset))
.expect("Failed to seek to file offset");
}
let content_length = response.content_length();
if content_length.is_none() {
warn!("recieved 0 length content from server");
return Err(ApplicationDownloadError::Communication(
RemoteAccessError::InvalidResponse(response.json().unwrap()),
));
}
let length = content_length.unwrap().try_into().unwrap();
if length != ctx.length {
return Err(ApplicationDownloadError::DownloadError);
for (i, raw_length) in lengths.split(",").enumerate() {
let length = raw_length.parse::<usize>().unwrap_or(0);
let Some(drop) = bucket.drops.get(i) else {
warn!("invalid number of Content-Lengths recieved: {}, {}", i, lengths);
return Err(ApplicationDownloadError::DownloadError);
};
if drop.length != length {
warn!(
"for {}, expected {}, got {} ({})",
drop.filename, drop.length, raw_length, length
);
return Err(ApplicationDownloadError::DownloadError);
}
}
let mut pipeline =
DropDownloadPipeline::new(response, destination, control_flag, progress, length);
DropDownloadPipeline::new(response, bucket.drops.clone(), control_flag, progress)
.map_err(|e| ApplicationDownloadError::IoError(e.kind()))?;
let completed = pipeline
.copy()
@ -192,23 +222,23 @@ pub fn download_game_chunk(
// If we complete the file, set the permissions (if on Linux)
#[cfg(unix)]
{
let permissions = Permissions::from_mode(ctx.permissions);
set_permissions(ctx.path.clone(), permissions).unwrap();
for drop in bucket.drops.iter() {
let permissions = Permissions::from_mode(drop.permissions);
set_permissions(drop.path.clone(), permissions)
.map_err(|e| ApplicationDownloadError::IoError(e.kind()))?;
}
}
let checksum = pipeline
let checksums = pipeline
.finish()
.map_err(|e| ApplicationDownloadError::IoError(e.kind()))?;
let res = hex::encode(checksum.0);
if res != ctx.checksum {
return Err(ApplicationDownloadError::Checksum);
for (index, drop) in bucket.drops.iter().enumerate() {
let res = hex::encode(**checksums.get(index).unwrap());
if res != drop.checksum {
return Err(ApplicationDownloadError::Checksum);
}
}
debug!(
"Successfully finished download #{}, copied {} bytes",
ctx.checksum, length
);
Ok(true)
}

8
src-tauri/src/games/downloads/drop_data.rs
View File

@ -76,14 +76,6 @@ impl DropData {
pub fn set_context(&self, context: String, state: bool) {
self.contexts.lock().unwrap().entry(context).insert_entry(state);
}
pub fn get_completed_contexts(&self) -> Vec<String> {
self.contexts
.lock()
.unwrap()
.iter()
.filter_map(|x| if *x.1 { Some(x.0.clone()) } else { None })
.collect()
}
pub fn get_contexts(&self) -> HashMap<String, bool> {
self.contexts.lock().unwrap().clone()
}

83
src-tauri/src/games/downloads/manifest.rs
View File

@ -2,6 +2,65 @@ use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
#[derive(Debug, Clone)]
// Drops go in buckets
pub struct DownloadDrop {
pub index: usize,
pub filename: String,
pub path: PathBuf,
pub start: usize,
pub length: usize,
pub checksum: String,
pub permissions: u32,
}
#[derive(Debug, Clone)]
pub struct DownloadBucket {
pub game_id: String,
pub version: String,
pub drops: Vec<DownloadDrop>,
}
#[derive(Deserialize)]
pub struct DownloadContext {
pub context: String,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ChunkBodyFile {
filename: String,
chunk_index: usize,
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ChunkBody {
pub context: String,
pub files: Vec<ChunkBodyFile>,
}
#[derive(Serialize)]
pub struct ManifestBody {
pub game: String,
pub version: String,
}
impl ChunkBody {
pub fn create(context: &DownloadContext, drops: &[DownloadDrop]) -> ChunkBody {
Self {
context: context.context.clone(),
files: drops
.iter()
.map(|e| ChunkBodyFile {
filename: e.filename.clone(),
chunk_index: e.index,
})
.collect(),
}
}
}
pub type DropManifest = HashMap<String, DropChunk>;
#[derive(Serialize, Deserialize, Debug, Clone, Ord, PartialOrd, Eq, PartialEq)]
#[serde(rename_all = "camelCase")]
@ -14,14 +73,26 @@ pub struct DropChunk {
}
#[derive(Serialize, Deserialize, Debug, Clone)]
pub struct DropDownloadContext {
pub file_name: String,
pub version: String,
pub struct DropValidateContext {
pub index: usize,
pub offset: u64,
pub game_id: String,
pub offset: usize,
pub path: PathBuf,
pub checksum: String,
pub length: usize,
pub permissions: u32,
}
impl From<DownloadBucket> for Vec<DropValidateContext> {
fn from(value: DownloadBucket) -> Self {
value
.drops
.into_iter()
.map(|e| DropValidateContext {
index: e.index,
offset: e.start,
path: e.path,
checksum: e.checksum,
length: e.length,
})
.collect()
}
}

18
src-tauri/src/games/downloads/validate.rs
View File

@ -7,24 +7,22 @@ use log::debug;
use md5::Context;
use crate::{
download_manager::
util::{
download_thread_control_flag::{DownloadThreadControl, DownloadThreadControlFlag},
progress_object::ProgressHandle,
}
,
download_manager::util::{
download_thread_control_flag::{DownloadThreadControl, DownloadThreadControlFlag},
progress_object::ProgressHandle,
},
error::application_download_error::ApplicationDownloadError,
games::downloads::manifest::DropDownloadContext,
games::downloads::manifest::DropValidateContext,
};
pub fn validate_game_chunk(
ctx: &DropDownloadContext,
ctx: &DropValidateContext,
control_flag: &DownloadThreadControl,
progress: ProgressHandle,
) -> Result<bool, ApplicationDownloadError> {
debug!(
"Starting chunk validation {}, {}, {} #{}",
ctx.file_name, ctx.index, ctx.offset, ctx.checksum
ctx.path.display(), ctx.index, ctx.offset, ctx.checksum
);
// If we're paused
if control_flag.get() == DownloadThreadControlFlag::Stop {
@ -38,7 +36,7 @@ pub fn validate_game_chunk(
if ctx.offset != 0 {
source
.seek(SeekFrom::Start(ctx.offset))
.seek(SeekFrom::Start(ctx.offset.try_into().unwrap()))
.expect("Failed to seek to file offset");
}