fix: potential download fixes

main/package.json

@@ -1,7 +1,7 @@
 {
   "name": "view",
   "private": true,
-  "version": "0.3.1",
+  "version": "0.3.2-dl",
   "type": "module",
   "scripts": {
     "build": "nuxt generate",

src-tauri/Cargo.lock

@@ -1284,7 +1284,7 @@ dependencies = [
 
 [[package]]
 name = "drop-app"
-version = "0.3.1"
+version = "0.3.2-dl"
 dependencies = [
  "atomic-instant-full",
  "bitcode",

src-tauri/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "drop-app"
-version = "0.3.1"
+version = "0.3.2-dl"
 description = "The client application for the open-source, self-hosted game distribution platform Drop"
 authors = ["Drop OSS"]
 edition = "2024"

src-tauri/src/download_manager/util/progress_object.rs

@@ -23,7 +23,7 @@ pub struct ProgressObject {
     //last_update: Arc<RwLock<Instant>>,
     last_update_time: Arc<AtomicInstant>,
     bytes_last_update: Arc<AtomicUsize>,
-    rolling: RollingProgressWindow<250>,
+    rolling: RollingProgressWindow<10>,
 }
 
 #[derive(Clone)]

src-tauri/src/games/downloads/download_logic.rs

@@ -7,14 +7,18 @@ use crate::error::drop_server_error::DropServerError;
 use crate::error::remote_access_error::RemoteAccessError;
 use crate::games::downloads::manifest::DropDownloadContext;
 use crate::remote::auth::generate_authorization_header;
-use log::{debug, warn};
+use http::response;
+use log::{debug, info, warn};
 use md5::{Context, Digest};
 use reqwest::blocking::{RequestBuilder, Response};
 
-use std::fs::{set_permissions, Permissions};
+use std::fs::{Permissions, set_permissions};
 use std::io::Read;
+use std::ops::Sub;
 #[cfg(unix)]
 use std::os::unix::fs::PermissionsExt;
+use std::thread;
+use std::time::Instant;
 use std::{
     fs::{File, OpenOptions},
     io::{self, BufWriter, Seek, SeekFrom, Write},
@@ -23,14 +27,21 @@ use std::{
 
 pub struct DropWriter<W: Write> {
     hasher: Context,
-    destination: W,
+    destination: BufWriter<W>,
+    progress: ProgressHandle,
 }
 impl DropWriter<File> {
-    fn new(path: PathBuf) -> Self {
+    fn new(path: PathBuf, progress: ProgressHandle) -> Self {
-        let destination = OpenOptions::new().write(true).create(true).truncate(false).open(&path).unwrap();
+        let destination = OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .open(&path)
+            .unwrap();
         Self {
-            destination,
+            destination: BufWriter::with_capacity(1 * 1024 * 1024, destination),
             hasher: Context::new(),
+            progress,
         }
     }
 
@@ -45,7 +56,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<()> {
@@ -64,60 +78,33 @@ pub struct DropDownloadPipeline<'a, R: Read, W: Write> {
     pub source: R,
     pub destination: DropWriter<W>,
     pub control_flag: &'a DownloadThreadControl,
-    pub progress: ProgressHandle,
     pub size: usize,
 }
+
+impl<'a> Seek for DropDownloadPipeline<'a, Response, File> {
+    fn seek(&mut self, pos: SeekFrom) -> io::Result<u64> {
+        self.destination.seek(pos)
+    }
+}
+
 impl<'a> DropDownloadPipeline<'a, Response, File> {
     fn new(
         source: Response,
-        destination: DropWriter<File>,
+        destination: PathBuf,
         control_flag: &'a DownloadThreadControl,
         progress: ProgressHandle,
         size: usize,
     ) -> Self {
         Self {
             source,
-            destination,
+            destination: DropWriter::new(destination, progress),
             control_flag,
-            progress,
             size,
         }
     }
 
     fn copy(&mut self) -> Result<bool, io::Error> {
-        let copy_buf_size = 512;
+        io::copy(&mut self.source, &mut self.destination).unwrap();
-        let mut copy_buf = vec![0; copy_buf_size];
-        let mut buf_writer = BufWriter::with_capacity(1024 * 1024, &mut self.destination);
-
-        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)
     }
@@ -139,11 +126,21 @@ pub fn download_game_chunk(
         progress.set(0);
         return Ok(false);
     }
+
+    let start = Instant::now();
+
+    debug!("started chunk {}", ctx.checksum);
+
+    let header = generate_authorization_header();
+    let header_time = start.elapsed();
+
     let response = request
-        .header("Authorization", generate_authorization_header())
+        .header("Authorization", header)
         .send()
         .map_err(|e| ApplicationDownloadError::Communication(e.into()))?;
 
+    let response_time = start.elapsed();
+
     if response.status() != 200 {
         debug!("chunk request got status code: {}", response.status());
         let raw_res = response.text().unwrap();
@@ -157,14 +154,6 @@ pub fn download_game_chunk(
         ));
     }
 
-    let mut destination = DropWriter::new(ctx.path.clone());
-
-    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");
@@ -179,8 +168,18 @@ pub fn download_game_chunk(
         return Err(ApplicationDownloadError::DownloadError);
     }
 
+    let pipeline_start = start.elapsed();
+
     let mut pipeline =
-        DropDownloadPipeline::new(response, destination, control_flag, progress, length);
+        DropDownloadPipeline::new(response, ctx.path.clone(), control_flag, progress, length);
+
+    if ctx.offset != 0 {
+        pipeline
+            .seek(SeekFrom::Start(ctx.offset))
+            .expect("Failed to seek to file offset");
+    }
+
+    let pipeline_setup = start.elapsed();
 
     let completed = pipeline
         .copy()
@@ -189,6 +188,8 @@ pub fn download_game_chunk(
         return Ok(false);
     }
 
+    let pipeline_finish = start.elapsed();
+
     // If we complete the file, set the permissions (if on Linux)
     #[cfg(unix)]
     {
@@ -200,15 +201,31 @@ pub fn download_game_chunk(
         .finish()
         .map_err(|e| ApplicationDownloadError::IoError(e.kind()))?;
 
+    let checksum_finish = start.elapsed();
+
     let res = hex::encode(checksum.0);
     if res != ctx.checksum {
         return Err(ApplicationDownloadError::Checksum);
     }
 
+    let header_update = header_time.as_millis();
+    let response_update = response_time.sub(header_time).as_millis();
+    let pipeline_start_update = pipeline_start.sub(response_time).as_millis();
+    let pipeline_setup_update = pipeline_setup.sub(pipeline_start).as_millis();
+    let pipeline_finish_update = pipeline_finish.sub(pipeline_setup).as_millis();
+    let checksum_update = checksum_finish.sub(pipeline_finish).as_millis();
+
     debug!(
-        "Successfully finished download #{}, copied {} bytes",
+        "\nheader: {}\nresponse: {}\npipeline start: {}\npipeline setup: {}\npipeline finish: {}\nchecksum finish: {}",
-        ctx.checksum, length
+        header_update,
+        response_update,
+        pipeline_start_update,
+        pipeline_setup_update,
+        pipeline_finish_update,
+        checksum_update
     );
 
+    debug!("finished chunk {}", ctx.checksum);
+
     Ok(true)
 }

src-tauri/src/lib.rs

@@ -1,5 +1,6 @@
 #![feature(fn_traits)]
 #![feature(duration_constructors)]
+#![feature(duration_millis_float)]
 #![deny(clippy::all)]
 
 mod database;

src-tauri/src/remote/cache.rs

@@ -11,7 +11,6 @@ use crate::{
 };
 use bitcode::{Decode, DecodeOwned, Encode};
 use http::{Response, header::CONTENT_TYPE, response::Builder as ResponseBuilder};
-use log::debug;
 
 #[macro_export]
 macro_rules! offline {
@@ -68,18 +67,9 @@ pub fn get_cached_object_db<D: DecodeOwned>(
     key: &str,
     db: &Database,
 ) -> Result<D, RemoteAccessError> {
-    let start = SystemTime::now();
     let bytes = read_sync(&db.cache_dir, key).map_err(RemoteAccessError::Cache)?;
-    let read = start.elapsed().unwrap();
     let data =
         bitcode::decode::<D>(&bytes).map_err(|e| RemoteAccessError::Cache(io::Error::other(e)))?;
-    let decode = start.elapsed().unwrap();
-    debug!(
-        "cache object took: r:{}, d:{}, b:{}",
-        read.as_millis(),
-        read.abs_diff(decode).as_millis(),
-        bytes.len()
-    );
     Ok(data)
 }
 #[derive(Encode, Decode)]

src-tauri/tauri.conf.json

@@ -1,7 +1,7 @@
 {
   "$schema": "https://schema.tauri.app/config/2.0.0",
   "productName": "Drop Desktop Client",
-  "version": "0.3.1",
+  "version": "0.3.2-dl",
   "identifier": "dev.drop.client",
   "build": {
     "beforeDevCommand": "yarn --cwd main dev --port 1432",