perf: run blocking data endpoints in a threadpool, not on the event loop

dataloom-backend/app/api/dependencies.py

@@ -13,6 +13,7 @@
 from app.services import auth_service
 from app.utils.logging import get_logger
 from app.utils.pandas_helpers import read_table_safe
+from app.utils.project_locks import project_write_lock
 from app.utils.rate_limiter import RateLimiter
 
 logger = get_logger(__name__)
@@ -166,7 +167,8 @@ def load_project_df(project: models.Project) -> pd.DataFrame:
     endpoint's handling. Shared by the profiling and visualization endpoints.
     """
     try:
-        return read_table_safe(project.file_path)
+        with project_write_lock(project.project_id):
+            return read_table_safe(project.file_path)
     except HTTPException as e:
         if e.status_code == 404:
             raise HTTPException(status_code=404, detail="Project data file not found") from e

dataloom-backend/app/api/endpoints/profiling.py

@@ -17,7 +17,7 @@
 
 
 @router.get("/{project_id}/profile/summary", response_model=schemas.DatasetSummaryResponse)
-async def get_dataset_summary(
+def get_dataset_summary(
     project_id: uuid.UUID,
     project: models.Project = Depends(get_project_or_404),
 ):
@@ -27,7 +27,7 @@ async def get_dataset_summary(
 
 
 @router.get("/{project_id}/profile/column", response_model=schemas.ColumnProfileResponse)
-async def get_column_profile(
+def get_column_profile(
     project_id: uuid.UUID,
     column_name: str = Query(..., description="Name of the column to profile"),
     project: models.Project = Depends(get_project_or_404),
@@ -47,7 +47,7 @@ async def get_column_profile(
 
 
 @router.get("/{project_id}/profile/columns", response_model=schemas.ColumnProfilesResponse)
-async def get_all_column_profiles(
+def get_all_column_profiles(
     project_id: uuid.UUID,
     project: models.Project = Depends(get_project_or_404),
 ):
@@ -62,7 +62,7 @@ async def get_all_column_profiles(
 
 
 @router.get("/{project_id}/profile/correlation", response_model=schemas.CorrelationResponse)
-async def get_correlation_matrix(
+def get_correlation_matrix(
     project_id: uuid.UUID,
     project: models.Project = Depends(get_project_or_404),
 ):

dataloom-backend/app/api/endpoints/projects.py

@@ -4,6 +4,7 @@
 """
 
 import os
+import shutil
 import tempfile
 import uuid
 from contextlib import suppress
@@ -14,6 +15,7 @@
 from sqlalchemy.exc import SQLAlchemyError
 from sqlmodel import Session
 from starlette.background import BackgroundTask
+from starlette.concurrency import run_in_threadpool
 
 from app import database, models, schemas
 from app.api.dependencies import get_current_user, get_project_or_404
@@ -35,6 +37,7 @@
 from app.utils.file_formats import TableWriteOptions, get_format, get_format_for_extension
 from app.utils.logging import get_logger
 from app.utils.pandas_helpers import dataframe_to_response, read_table_safe, save_table_safe
+from app.utils.project_locks import project_write_lock
 from app.utils.security import validate_upload_file
 
 logger = get_logger(__name__)
@@ -63,9 +66,9 @@ async def upload_project(
     logger.info("Upload request: project=%s, file=%s", projectName, file.filename)
     await validate_upload_file(file)
 
-    original_path, copy_path = store_upload(file)
+    original_path, copy_path = await run_in_threadpool(store_upload, file)
     try:
-        df = read_table_safe(original_path)
+        df = await run_in_threadpool(read_table_safe, original_path)
     except HTTPException as e:
         # The just-uploaded file could not be parsed — that's a bad client file,
         # not a server fault. Discard the orphaned files and report a clean 400.
@@ -113,13 +116,14 @@ def list_projects(
 
 
 @router.get("/get/{project_id}", response_model=schemas.ProjectResponse)
-async def get_project_details(
+def get_project_details(
     page: int = 1,
     pageSize: int = 50,
     project: models.Project = Depends(get_project_or_404),
 ):
     """Fetch full project details including all rows and columns."""
-    df = read_table_safe(project.file_path)
+    with project_write_lock(project.project_id):
+        df = read_table_safe(project.file_path)
 
     total_rows = len(df)
     total_pages = (total_rows + pageSize - 1) // pageSize
@@ -179,7 +183,7 @@ async def rename_project_endpoint(
 
 
 @router.post("/{project_id}/save", response_model=schemas.ProjectResponse)
-async def save_project(
+def save_project(
     project_id: uuid.UUID,
     commit_message: str,
     db: Session = Depends(database.get_db),
@@ -191,23 +195,24 @@ async def save_project(
     checkpoint creation should preserve that current dataset and only update the
     checkpoint/log metadata for pending actions.
     """
-    original_path = get_original_path(project.file_path)
-    if Path(project.file_path).resolve() == original_path.resolve():
-        logger.error(
-            "Project working copy unexpectedly points at original file: id=%s working_copy=%s original=%s",
-            project_id,
-            project.file_path,
-            original_path,
-        )
-        raise HTTPException(
-            status_code=500,
-            detail=f"Project {project_id} working copy is misconfigured; please retry or contact support.",
-        )
+    with project_write_lock(project_id):
+        original_path = get_original_path(project.file_path)
+        if Path(project.file_path).resolve() == original_path.resolve():
+            logger.error(
+                "Project working copy unexpectedly points at original file: id=%s working_copy=%s original=%s",
+                project_id,
+                project.file_path,
+                original_path,
+            )
+            raise HTTPException(
+                status_code=500,
+                detail=f"Project {project_id} working copy is misconfigured; please retry or contact support.",
+            )
 
-    df = read_table_safe(project.file_path)
+        df = read_table_safe(project.file_path)
 
-    # Create checkpoint (marks logs as applied)
-    checkpoint = create_checkpoint(db, project_id, commit_message)
+        # Create checkpoint (marks logs as applied)
+        checkpoint = create_checkpoint(db, project_id, commit_message)
 
     total_rows = len(df)
     resp = dataframe_to_response(df)
@@ -225,7 +230,7 @@ async def save_project(
 
 
 @router.post("/{project_id}/revert", response_model=schemas.ProjectResponse)
-async def revert_to_checkpoint(
+def revert_to_checkpoint(
     project_id: uuid.UUID,
     checkpoint_id: uuid.UUID = None,
     db: Session = Depends(database.get_db),
@@ -237,6 +242,11 @@ async def revert_to_checkpoint(
     that checkpoint onto the original file. When None, reverts to the original
     uploaded state.
     """
+    with project_write_lock(project_id):
+        return _revert_to_checkpoint(project_id, checkpoint_id, db, project)
+
+
+def _revert_to_checkpoint(project_id, checkpoint_id, db, project):
     original_path = get_original_path(project.file_path)
     df = read_table_safe(original_path)
 
@@ -308,7 +318,7 @@ async def revert_to_checkpoint(
 
 
 @router.get("/{project_id}/export")
-async def export_project(
+def export_project(
     fmt: str | None = Query(default=None, alias="format"),
     delimiter: str | None = Query(default=None),
     include_header: bool = Query(default=True),
@@ -350,20 +360,20 @@ async def export_project(
         encoding=encoding,
     )
 
-    # Native export — no conversion and no delimited options means we can stream
-    # the working copy directly.
-    if target_fmt.extension == source_fmt.extension and not write_options.has_options():
-        return FileResponse(
-            project.file_path,
-            media_type=target_fmt.media_type,
-            filename=f"{project.name}{target_fmt.extension}",
-        )
+    native = target_fmt.extension == source_fmt.extension and not write_options.has_options()
 
-    df = read_table_safe(project.file_path)
+    # Snapshot under the project lock. FileResponse streams after we return,
+    # so serving the live working copy could tear if a writer starts mid-download.
     with tempfile.NamedTemporaryFile(suffix=target_fmt.extension, delete=False) as tmp:
         tmp_path = tmp.name
     try:
-        save_table_safe(df, Path(tmp_path), write_options)
+        with project_write_lock(project.project_id):
+            if native:
+                shutil.copyfile(project.file_path, tmp_path)
+            else:
+                df = read_table_safe(project.file_path)
+        if not native:
+            save_table_safe(df, Path(tmp_path), write_options)
         return FileResponse(
             tmp_path,
             media_type=target_fmt.media_type,
@@ -417,7 +427,7 @@ async def delete_project_endpoint(
 
 
 @router.post("/{project_id}/undo", response_model=schemas.ProjectResponse)
-async def undo_last_transformation(
+def undo_last_transformation(
     project_id: uuid.UUID,
     project: models.Project = Depends(get_project_or_404),
     db: Session = Depends(database.get_db),
@@ -427,6 +437,11 @@ async def undo_last_transformation(
     Removes the last change log entry and rebuilds the working copy
     by replaying all remaining logs onto the original file.
     """
+    with project_write_lock(project_id):
+        return _undo_last_transformation(project_id, project, db)
+
+
+def _undo_last_transformation(project_id, project, db):
     last_log = get_last_change_log(db, project_id)
     if not last_log:
         raise HTTPException(status_code=404, detail="No transformations to undo")

dataloom-backend/app/api/endpoints/transformations.py

@@ -16,6 +16,7 @@
 from app.services.project_service import log_transformations_or_restore
 from app.utils.logging import get_logger
 from app.utils.pandas_helpers import dataframe_to_response, read_table_safe, save_table_safe
+from app.utils.project_locks import project_write_lock
 from app.utils.security import safe_transformation_error_detail
 
 logger = get_logger(__name__)
@@ -65,7 +66,7 @@ def _dispatch_transform(df, transformation_input):
 
 
 @router.post("/{project_id}/transform", response_model=schemas.BasicQueryResponse)
-async def transform_project(
+def transform_project(
     project_id: uuid.UUID,
     transformation_input: schemas.TransformationInput,
     preview: bool = Query(False, description="If true, return transformation data without saving."),
@@ -77,10 +78,13 @@ async def transform_project(
     """Apply a transformation to a project.
 
     Routes to the appropriate internal handler based on operation_type.
+    Preview still reads ``project.file_path``, so it shares the same lock.
     """
-    # Keep an explicit local for consistency across dispatch, persistence and
-    # logging paths. Use a defensive fallback so exception logging never
-    # introduces a secondary NameError.
+    with project_write_lock(project_id):
+        return _transform_project(project_id, transformation_input, preview, page, page_size, db, project)
+
+
+def _transform_project(project_id, transformation_input, preview, page, page_size, db, project):
     operation_type = getattr(transformation_input, "operation_type", "<unknown>")
 
     try:

dataloom-backend/app/utils/project_locks.py

@@ -0,0 +1,33 @@
+"""Per-project synchronization for file and change-log mutations."""
+
+import threading
+import uuid
+from collections.abc import Iterator
+from contextlib import contextmanager
+
+_registry_lock = threading.Lock()
+_project_locks: dict[uuid.UUID, tuple[threading.Lock, int]] = {}
+
+
+@contextmanager
+def project_write_lock(project_id: uuid.UUID) -> Iterator[None]:
+    """Serialize reads and writes of one project's working copy.
+
+    The same lock covers writers and readers of ``project.file_path`` so a
+    threadpool reader cannot observe a torn in-place write. Independent
+    projects stay concurrent.
+    """
+    with _registry_lock:
+        lock, users = _project_locks.get(project_id, (threading.Lock(), 0))
+        _project_locks[project_id] = (lock, users + 1)
+
+    try:
+        with lock:
+            yield
+    finally:
+        with _registry_lock:
+            current_lock, users = _project_locks[project_id]
+            if users == 1:
+                del _project_locks[project_id]
+            else:
+                _project_locks[project_id] = (current_lock, users - 1)