⚡ convert snapshot_and_load file copies to tokio::fs::copy

services/receiver/src/rd_poll.rs

@@ -48,19 +48,42 @@ fn signature(dir: &Path) -> Option<FileSignature> {
 /// Copy each active RD file into a fresh temp directory and parse the copies. Copying
 /// first mitigates sharing violations / torn reads while RD writes the
 /// originals. Any copy or parse error is returned so the caller keeps last good.
-fn snapshot_and_load(dir: &Path) -> Result<RdImport, RdError> {
-    let files = rd_dbf::resolve_rd_files(dir)?;
-    let tmp = tempfile::tempdir().map_err(|e| RdError::Io {
-        file: "<tempdir>".to_owned(),
-        source: e,
-    })?;
-    for file in files {
-        std::fs::copy(dir.join(file), tmp.path().join(file)).map_err(|e| RdError::Io {
-            file: file.to_owned(),
+async fn snapshot_and_load(dir: &Path) -> Result<RdImport, RdError> {
+    let dir_owned = dir.to_path_buf();
+    let (files, tmp) = tokio::task::spawn_blocking(move || {
+        let files = rd_dbf::resolve_rd_files(&dir_owned)?;
+        let tmp = tempfile::tempdir().map_err(|e| RdError::Io {
+            file: "<tempdir>".to_owned(),
             source: e,
         })?;
+        Ok::<_, RdError>((files, tmp))
+    })
+    .await
+    .map_err(|e| RdError::Io {
+        file: "<spawn_blocking>".to_owned(),
+        source: std::io::Error::other(e),
+    })??;
+
+    for file in files {
+        tokio::fs::copy(dir.join(file), tmp.path().join(file))
+            .await
+            .map_err(|e| RdError::Io {
+                file: file.to_owned(),
+                source: e,
+            })?;
     }
-    rd_dbf::load_from_dir(tmp.path())
+
+    let tmp_path = tmp.path().to_path_buf();
+    let import = tokio::task::spawn_blocking(move || rd_dbf::load_from_dir(&tmp_path))
+        .await
+        .map_err(|e| RdError::Io {
+            file: "<spawn_blocking>".to_owned(),
+            source: std::io::Error::other(e),
+        })??;
+
+    tokio::task::spawn_blocking(move || drop(tmp)).await.ok();
+
+    Ok(import)
 }
 
 /// Run one poll pass against `dir`.
@@ -81,18 +104,15 @@ pub async fn poll_once(
         return PollOutcome::Skipped;
     }
 
-    let dir_owned = dir.to_path_buf();
-    let loaded = tokio::task::spawn_blocking(move || snapshot_and_load(&dir_owned)).await;
-    match loaded {
-        Ok(Ok(import)) => match crate::control_api::apply_rd_import(state, import).await {
+    match snapshot_and_load(dir).await {
+        Ok(import) => match crate::control_api::apply_rd_import(state, import).await {
             Ok(summary) => {
                 *last = sig;
                 PollOutcome::Imported(summary.resolvable_chips)
             }
             Err(e) => PollOutcome::Failed(e.to_string()),
         },
-        Ok(Err(e)) => PollOutcome::Failed(e.to_string()),
-        Err(e) => PollOutcome::Failed(format!("import task failed: {e}")),
+        Err(e) => PollOutcome::Failed(e.to_string()),
     }
 }
 

services/receiver/tests/bench_rd_poll.rs

@@ -0,0 +1,46 @@
+use receiver::control_api::AppState;
+use receiver::db::Db;
+use receiver::rd_dbf::RD_FILES;
+use receiver::rd_poll::poll_once;
+use std::path::{Path, PathBuf};
+use std::time::Instant;
+
+fn fixtures_dir() -> PathBuf {
+    PathBuf::from(env!("CARGO_MANIFEST_DIR"))
+        .join("tests")
+        .join("fixtures")
+        .join("rd")
+}
+
+fn copy_fixtures_to(dir: &Path) {
+    for file in RD_FILES {
+        std::fs::copy(fixtures_dir().join(file), dir.join(file)).unwrap();
+    }
+}
+
+#[tokio::test]
+#[ignore]
+async fn bench_rd_poll_once() {
+    let (state, _rx) = AppState::new(Db::open_in_memory().unwrap(), "recv".to_owned());
+    let tmp = tempfile::tempdir().unwrap();
+    copy_fixtures_to(tmp.path());
+
+    // Warmup
+    let mut last = None;
+    poll_once(&state, tmp.path(), &mut last).await;
+
+    let iterations = 100;
+    let start = Instant::now();
+    for _ in 0..iterations {
+        let mut last = None; // Force snapshot_and_load each iteration
+        poll_once(&state, tmp.path(), &mut last).await;
+    }
+    let elapsed = start.elapsed();
+    let avg = elapsed / iterations;
+
+    println!("\n==================================================");
+    println!("BENCHMARK: poll_once snapshot_and_load ({iterations} iterations):");
+    println!("Total time: {:?}", elapsed);
+    println!("Average time per poll: {:?}", avg);
+    println!("==================================================\n");
+}