perf(export): run export cores on the blocking pool

crates/dbx-core/src/export_runtime.rs

@@ -0,0 +1,89 @@
+//! Execution context for long-running export tasks.
+
+use std::future::Future;
+
+/// Spawns a long-running export task that interleaves async database fetches
+/// with synchronous row formatting and buffered disk writes.
+///
+/// Running such a task directly on a tokio worker (`tokio::spawn`) blocks that
+/// worker for the whole synchronous write slice of every page, so a multi-GB
+/// export stalls every concurrent command sharing the runtime. This helper
+/// instead runs the task on the blocking pool and drives it with
+/// [`tokio::runtime::Handle::block_on`]: the blocking thread owns the
+/// synchronous formatting and writes while async fetches still execute on the
+/// regular workers. The blocking pool grows on demand, so concurrent exports
+/// and other `spawn_blocking` users do not starve each other.
+pub fn spawn_export_task(task: impl Future<Output = ()> + Send + 'static) {
+    let handle = tokio::runtime::Handle::current();
+    tokio::task::spawn_blocking(move || handle.block_on(task));
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicBool, Ordering};
+    use std::sync::Arc;
+    use std::time::Duration;
+
+    /// The task must run to completion from a blocking-pool thread: async
+    /// timers, `tokio::fs`, and nested `spawn_blocking` all need the runtime
+    /// context that `Handle::block_on` provides there.
+    #[tokio::test]
+    async fn export_task_runs_async_and_blocking_work_to_completion() {
+        let (tx, rx) = tokio::sync::oneshot::channel::<u64>();
+        super::spawn_export_task(async move {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+            let nested = tokio::task::spawn_blocking(|| 40u64 + 2).await.unwrap_or(0);
+            tx.send(nested).expect("receiver dropped");
+        });
+        let value = tokio::time::timeout(Duration::from_secs(10), rx)
+            .await
+            .expect("export task did not finish")
+            .expect("export task dropped its sender");
+        assert_eq!(value, 42);
+    }
+
+    /// The reason exports run on the blocking pool: a synchronous write slice
+    /// inside the task must not occupy the (single) async worker, so unrelated
+    /// async tasks keep making progress while the export is writing to disk.
+    /// If the task were spawned with `tokio::spawn` instead, the 600ms
+    /// blocking slice below would pin the only worker; the quick task could
+    /// not even observe the slice flag until the slice ended, and the total
+    /// elapsed time would reach the full slice duration.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    async fn export_blocking_slice_does_not_starve_async_tasks() {
+        const SLICE: Duration = Duration::from_millis(600);
+
+        // Set right before the export task enters its synchronous slice.
+        let in_slice = Arc::new(AtomicBool::new(false));
+        let (quick_tx, quick_rx) = tokio::sync::oneshot::channel::<()>();
+        tokio::spawn({
+            let watch = in_slice.clone();
+            async move {
+                while !watch.load(Ordering::SeqCst) {
+                    tokio::time::sleep(Duration::from_millis(1)).await;
+                }
+                tokio::time::sleep(Duration::from_millis(10)).await;
+                let _ = quick_tx.send(());
+            }
+        });
+
+        let started = std::time::Instant::now();
+        let signal = in_slice.clone();
+        super::spawn_export_task(async move {
+            tokio::task::yield_now().await;
+            signal.store(true, Ordering::SeqCst);
+            // Simulates the synchronous row-format + buffered disk write slice.
+            std::thread::sleep(SLICE);
+        });
+
+        tokio::time::timeout(Duration::from_secs(5), quick_rx)
+            .await
+            .expect("quick async task did not finish")
+            .expect("quick task dropped its sender");
+        let elapsed = started.elapsed();
+        assert!(
+            elapsed < SLICE - Duration::from_millis(100),
+            "async tasks were starved for {elapsed:?} while the export wrote to disk"
+        );
+    }
+}

crates/dbx-core/src/lib.rs

@@ -46,6 +46,7 @@ pub mod dml_binding;
 pub mod docs;
 pub mod document_ops;
 pub mod driver_runtime;
+pub mod export_runtime;
 pub mod external;
 pub mod hbase_ops;
 pub mod history;

crates/dbx-web/src/routes/database_export.rs

@@ -67,7 +67,9 @@ pub async fn start_database_export(
     let cancelled = Arc::new(AtomicBool::new(false));
     let cancelled_progress = cancelled.clone();
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let result = database_export::export_database_sql_core(&app, &req, |progress| {
             if matches!(progress.status, ExportStatus::Cancelled) {
                 cancelled_progress.store(true, Ordering::SeqCst);

crates/dbx-web/src/routes/query_result_export.rs

@@ -69,7 +69,9 @@ pub async fn start_query_result_export(
     let cancelled = Arc::new(AtomicBool::new(false));
     let cancelled_progress = cancelled.clone();
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let execution_id = req.execution_id.clone().filter(|id| !id.trim().is_empty());
         let registered_query = execution_id.as_ref().map(|id| {
             app.running_queries.register_task(

crates/dbx-web/src/routes/table_export.rs

@@ -71,7 +71,9 @@ pub async fn start_table_export(
     let cancelled = Arc::new(AtomicBool::new(false));
     let cancelled_progress = cancelled.clone();
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let result = table_export::export_table_data_core(&app, &req, |progress| {
             if matches!(progress.status, ExportStatus::Cancelled) {
                 cancelled_progress.store(true, Ordering::SeqCst);

src-tauri/src/commands/database_export.rs

@@ -34,7 +34,9 @@ pub async fn export_database_sql(
     let state = state.inner().clone();
     let export_id = request.export_id.clone();
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let result = dbx_core::database_export::export_database_sql_core(&state, &request, |progress| {
             emit_progress(&app, progress)
         })

src-tauri/src/commands/query_result_export.rs

@@ -269,7 +269,9 @@ pub async fn start_query_result_export(
     let target = StagedExportTarget::new(&request.file_path)?;
     request.file_path = target.path_string()?;
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let execution_id = request.execution_id.clone().filter(|id| !id.trim().is_empty());
         let registered_query = execution_id.as_ref().map(|id| {
             state.running_queries.register_task(

src-tauri/src/commands/table_export.rs

@@ -29,7 +29,9 @@ pub async fn start_table_export(
     let export_id = request.export_id.clone();
     let file_path = request.file_path.clone();
 
-    tokio::spawn(async move {
+    // Exports interleave async fetches with synchronous row formatting and
+    // buffered disk writes; run them off the async workers (see spawn_export_task).
+    dbx_core::export_runtime::spawn_export_task(async move {
         let cancelled = Arc::new(AtomicBool::new(false));
         let cancelled_progress = cancelled.clone();
         let result = dbx_core::table_export::export_table_data_core(&state, &request, |progress| {