Some fixes and optimizations

Author: rapuckett

Cargo.lock

@@ -42,6 +42,7 @@ async-trait = "0.1.83"
 tmpdir = "1.0.0"
 num_cpus = "1.16.0"
 tokio-stream = "0.1.16"
+bytes = "1.7.2"
 
 [target.'cfg(target_os = "android")'.dependencies]
 jni = "0.21.1"

Cargo.toml

@@ -1,4 +1,4 @@
-use actix_web::{HttpResponse, ResponseError};
+use actix_web::{HttpResponse, ResponseError, error::BlockingError};
 use anyhow::anyhow;
 use anyhow::Error as AnyhowError;
 use base64_url::base64;
@@ -66,5 +66,11 @@ impl From<ErrReport> for AppError {
     }
 }
 
+impl From<BlockingError> for AppError {
+    fn from(error: BlockingError) -> Self {
+        AppError(AnyhowError::new(error).context("An error occurred during a blocking operation"))
+    }
+}
+
 // Shortcut for Result<T, AppError>
 pub type AppResult<T> = std::result::Result<T, AppError>;

src/error.rs

@@ -23,6 +23,8 @@ pub fn scope() -> actix_web::Scope {
         )
 }
 
+// This doesn't seem to be the way to delete a group.
+//
 #[delete("")]
 async fn delete_group(group_id: web::Path<String>) -> AppResult<impl Responder> {
     let backend = get_backend().await?;

src/groups.rs

@@ -1,4 +1,6 @@
+#[cfg(target_os = "android")]
 use std::ffi::CString;
+
 #[cfg(target_os = "android")]
 use std::os::raw::{c_char, c_int};
 

src/logging.rs

@@ -1,15 +1,15 @@
 use crate::constants::TAG;
-use crate::error::AppResult;
-use crate::logging::android_log;
+use crate::error::{AppError, AppResult};
 use crate::models::{GroupRepoMediaPath, GroupRepoPath};
 use crate::server::server::get_backend;
 use crate::utils::create_veilid_cryptokey_from_base64;
-use crate::{log_debug, log_info};
-use actix_web::web::Bytes;
-use actix_web::{delete, get, post, web, HttpResponse, Responder, Scope};
+use crate::log_info;
+use actix_web::{delete, get, post, web, HttpResponse, Responder, Scope, http::header, error::BlockingError};
+use bytes::{BytesMut, Bytes};
+use futures::Stream;
 use futures::StreamExt;
-use save_dweb_backend::common::DHTEntity;
 use serde_json::json;
+use std::io;
 
 pub fn scope() -> Scope {
     web::scope("/media")
@@ -19,6 +19,27 @@ pub fn scope() -> Scope {
         .service(download_file)
 }
 
+pub fn from_blocking<T>(result: Result<T, BlockingError>) -> AppResult<T> {
+    result.map_err(AppError::from)
+}
+
+async fn handle_file_stream(mut file_data: impl Stream<Item = Result<Bytes, io::Error>> + Unpin) -> AppResult<(usize, Bytes)> {
+    let mut buffer = BytesMut::new();
+    let mut length = 0;
+
+    while let Some(chunk_result) = file_data.next().await {
+        let chunk = chunk_result.map_err(|e| AppError(anyhow::Error::new(e)))?;
+        buffer.extend_from_slice(&chunk);
+        length += chunk.len();
+    }
+
+    let final_buffer = web::block(move || {
+        buffer.freeze()
+    }).await?;
+
+    Ok((length, final_buffer))
+}
+
 #[get("")]
 async fn list_files(path: web::Path<GroupRepoPath>) -> AppResult<impl Responder> {
     let path_params = path.into_inner();
@@ -55,8 +76,7 @@ async fn list_files(path: web::Path<GroupRepoPath>) -> AppResult<impl Responder>
             "is_downloaded": is_downloaded
         }));
     }
-
-    Ok(HttpResponse::Ok().json(files_with_status))
+    Ok(HttpResponse::Ok().json(json!({ "files": files_with_status })))
 }
 
 #[get("/{file_name}")]
@@ -93,13 +113,14 @@ async fn download_file(path: web::Path<GroupRepoMediaPath>) -> AppResult<impl Re
     // Trigger file download from peers using the hash
     let file_data = repo
         .get_file_stream(file_name)
-        .await
-        .map_err(|e| anyhow::anyhow!("Failed to download file from peers: {}", e))?;
+        .await?;
+
+    let (content_length, buffered_data) = handle_file_stream(file_data).await?;
 
-    // Return the file data as a binary response
     Ok(HttpResponse::Ok()
         .content_type("application/octet-stream")
-        .streaming(file_data))
+        .insert_header((header::CONTENT_LENGTH, content_length))
+        .body(buffered_data))
 }
 
 #[delete("/{file_name}")]
@@ -175,6 +196,7 @@ async fn upload_file(
         .map_err(|e| anyhow::anyhow!("Failed to upload file: {}", e))?;
 
     Ok(HttpResponse::Ok().json(json!({
+        "name": file_name,
         "updated_collection_hash": updated_collection_hash,
     })))
 }

src/media.rs

@@ -103,17 +103,28 @@ async fn create_repo(
 }
 
 async fn get_snowbird_repos(group: &Group) -> Result<Vec<SnowbirdRepo>, AppError> {
-    log_debug!(TAG, "start");
+    log_debug!(TAG, "Getting Snowbird Repos");
+
+    let locked_repos = group.repos.lock().await;
+    let repos: Vec<_> = locked_repos.values().cloned().collect();
 
-    let repo_ids = group.list_repos().await;
     let mut snowbird_repos = Vec::new();
 
-    for id in repo_ids {
-        log_debug!(TAG, "Repo ID {}", id);
-        let repo = group.get_repo(&id).await?;
+    for repo in repos {
+        log_debug!(TAG, "Repo ID {}", repo.id());
         let snowbird_repo = SnowbirdRepo::async_from(repo).await;
         snowbird_repos.push(snowbird_repo);
     }
 
+    // let repo_ids = group.list_repos().await;
+    // let mut snowbird_repos = Vec::new();
+
+    // for id in repo_ids {
+    //     log_debug!(TAG, "Repo ID {}", id);
+    //     let repo = group.get_repo(&id).await?;
+    //     let snowbird_repo = SnowbirdRepo::async_from(repo).await;
+    //     snowbird_repos.push(snowbird_repo);
+    // }
+
     Ok(snowbird_repos)
 }

src/repos.rs

@@ -92,11 +92,14 @@ pub mod server {
 
     fn get_optimal_worker_count() -> usize {
         let cpu_count = num_cpus::get();
-        let worker_count = cmp::max(1, cmp::min(cpu_count / 2, 4));
+        //let worker_count = cmp::max(1, cmp::min(cpu_count / 2, 4));
 
-        log_debug!(TAG, "Detected {} CPUs, using {} workers", cpu_count, worker_count);
+        log_debug!(TAG, "Detected {} CPUs", cpu_count);
 
-        worker_count
+        // This whole thing was an attempt at optimization, but since
+        // we're only ever handling one request at a time let's keep
+        // things lightweight for now.
+        1
     }
 
     pub async fn start(backend_base_directory: &str, server_socket_path: &str) -> anyhow::Result<()> {