fix: add concurrency limit to prevent crash on large graphs

rsapi-impl/src/error.rs

@@ -14,6 +14,8 @@ pub enum Error {
     InvalidArg,
     #[error("cancelled")]
     Cancelled,
+    #[error("task join error: {0}")]
+    JoinError(#[from] tokio::task::JoinError),
 }
 
 pub type Result<T> = std::result::Result<T, Error>;

rsapi-impl/src/graph.rs

@@ -14,6 +14,11 @@ use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::sync::watch;
 
 use futures::prelude::*;
+use futures::stream::{self, StreamExt};
+
+/// Maximum concurrent file operations to prevent resource exhaustion on large graphs.
+/// See: https://github.com/logseq/logseq/issues/12331
+const MAX_CONCURRENT_FILE_OPS: usize = 50;
 use lsq_encryption::md5_hexdigest;
 use sync::SyncClient;
 use unicode_normalization::UnicodeNormalization;
@@ -203,11 +208,12 @@ impl Graph {
                 })
                 .map(|p| self.get_file_meta(&base_path_ref, p))
         };
-        Ok(future::join_all(futs)
+        Ok(stream::iter(futs)
+            .buffer_unordered(MAX_CONCURRENT_FILE_OPS)
+            .filter_map(|r| async { r.ok() })
+            .collect::<Vec<_>>()
             .await
-            .into_iter()
-            .filter(Result::is_ok)
-            .map(Result::unwrap))
+            .into_iter())
     }
 
     pub async fn rename_local_file<P: AsRef<Path>, S0: AsRef<str>, S1: AsRef<str>>(
@@ -367,7 +373,7 @@ impl Graph {
         }
 
         tokio::select! {
-            ret = future::join_all(tasks) => {
+            ret = stream::iter(tasks).buffer_unordered(MAX_CONCURRENT_FILE_OPS).collect::<Vec<_>>() => {
                 ret.into_iter().filter_map(|f| f.transpose()).collect()
             }
             _ = cancel_notification.changed() => {
@@ -453,7 +459,7 @@ impl Graph {
         }
 
         tokio::select! {
-            ret = future::join_all(tasks) => {
+            ret = stream::iter(tasks).buffer_unordered(MAX_CONCURRENT_FILE_OPS).collect::<Vec<_>>() => {
                 ret.into_iter().collect::<Result<Vec<_>>>().map(|_| ())
             }
             _ = cancel_notification.changed() => {
@@ -487,6 +493,11 @@ impl Graph {
 
         let mut tasks = vec![];
         let mut page_files = vec![];
+
+        // Clone keys for use in spawn_blocking (requires 'static lifetime)
+        let age_public_key = Arc::new(self.age_public_key.clone());
+        let fname_encryption_key = Arc::new(self.fname_encryption_key);
+
         for file_path in file_paths {
             // move in variables
             let client = client.clone();
@@ -496,6 +507,8 @@ impl Graph {
             }
 
             let full_file_path = base_path.join(&file_path);
+            let age_public_key = age_public_key.clone();
+            let fname_encryption_key = fname_encryption_key.clone();
 
             let progress_callback = {
                 let file_path = file_path.clone();
@@ -530,26 +543,48 @@ impl Graph {
             };
             tasks.push(async move {
                 let content = fs::read(full_file_path).await?;
-                // stage 1.1: md5 metadata
-                let md5checksum = md5_hexdigest(&content);
-                // stage 1.2: encryption
-                let encrypted = self.encrypt_content(&content)?;
+
+                // Move CPU-bound work (MD5 + encryption) to blocking thread pool
+                let file_path_clone = file_path.clone();
+                let (md5checksum, encrypted, encrypted_file_path) =
+                    tokio::task::spawn_blocking(move || {
+                        // stage 1.1: md5 metadata
+                        let md5checksum = md5_hexdigest(&content);
+
+                        // stage 1.2: encryption
+                        let encrypted = if content.starts_with(b"-----BEGIN AGE ENCRYPTED FILE-----")
+                            || content.starts_with(b"age-encryption.org/v1\n")
+                        {
+                            content.into()
+                        } else {
+                            lsq_encryption::encrypt_with_x25519(&age_public_key, &content, false)?
+                                .to_vec()
+                                .into()
+                        };
+
+                        // stage 1.3: encrypt filename
+                        let encrypted_file_path =
+                            lsq_encryption::encrypt_filename(&file_path_clone, &fname_encryption_key)?;
+
+                        Result::Ok((md5checksum, encrypted, encrypted_file_path))
+                    })
+                    .await??;
+
+                let encrypted: std::borrow::Cow<'_, [u8]> = encrypted;
                 if encrypted.len() > 10 * 1024 * 1024 {
                     log::warn!(
-                        "large file {:?} size: {:.2}MiB encrypted: {:.2}MiB",
+                        "large file {:?} encrypted: {:.2}MiB",
                         file_path,
-                        content.len() as f64 / (1024.0 * 1024.0),
                         encrypted.len() as f64 / (1024.0 * 1024.0)
                     );
                 }
                 let remote_temp_url = client.upload_tempfile(encrypted, progress_callback).await?;
-                let encrypted_file_path = self.encrypt_filename(&file_path)?;
                 Result::Ok((encrypted_file_path, remote_temp_url, md5checksum))
             });
         }
 
         tokio::select! {
-            task_results = future::join_all(tasks) => {
+            task_results = stream::iter(tasks).buffer_unordered(MAX_CONCURRENT_FILE_OPS).collect::<Vec<_>>() => {
                 let temp_remote_files = task_results.into_iter().collect::<Result<Vec<_>>>()?;
                 let update = client.update_files(temp_remote_files).await?;
                 for path in page_files {

sync/src/helpers.rs

@@ -33,16 +33,15 @@ impl Stream for ProgressedBytesStream {
             return Poll::Ready(None);
         }
 
-        let mut buf = [0; 8 * 1024];
-        let mut len = 0;
-        // TODO: optimize
-        while len < buf.len() && self.offset < self.inner.len() {
-            buf[len] = self.inner[self.offset];
-            len += 1;
-            self.offset += 1;
-        }
+        // Use 64KB chunks for better throughput
+        const CHUNK_SIZE: usize = 64 * 1024;
+
+        let remaining = self.inner.len() - self.offset;
+        let chunk_len = remaining.min(CHUNK_SIZE);
+        let chunk = Bytes::copy_from_slice(&self.inner[self.offset..self.offset + chunk_len]);
+        self.offset += chunk_len;
 
         (self.callback)(self.offset, self.inner.len());
-        Poll::Ready(Some(Ok(Bytes::from(buf[..len].to_vec()))))
+        Poll::Ready(Some(Ok(chunk)))
     }
 }