fix: Make sync operations truly non-blocking with thread pool (#309)

Signed-off-by: phernandez <[email protected]>
Co-authored-by: Claude <[email protected]>

Author: phernandez

src/basic_memory/config.py

@@ -82,6 +82,12 @@ class BasicMemoryConfig(BaseSettings):
         description="Whether to sync changes in real time. default (True)",
     )
 
+    sync_thread_pool_size: int = Field(
+        default=4,
+        description="Size of thread pool for file I/O operations in sync service",
+        gt=0,
+    )
+
     kebab_filenames: bool = Field(
         default=False,
         description="Format for generated filenames. False preserves spaces and special chars, True converts them to hyphens for consistency with permalinks",

src/basic_memory/services/initialization.py

@@ -101,31 +101,38 @@ async def initialize_file_sync(
     # Get active projects
     active_projects = await project_repository.get_active_projects()
 
-    # First, sync all projects sequentially
-    for project in active_projects:
+    # Start sync for all projects as background tasks (non-blocking)
+    async def sync_project_background(project):
+        """Sync a single project in the background."""
         # avoid circular imports
         from basic_memory.cli.commands.sync import get_sync_service
 
-        logger.info(f"Starting sync for project: {project.name}")
-        sync_service = await get_sync_service(project)
-        sync_dir = Path(project.path)
-
+        logger.info(f"Starting background sync for project: {project.name}")
         try:
+            sync_service = await get_sync_service(project)
+            sync_dir = Path(project.path)
             await sync_service.sync(sync_dir, project_name=project.name)
-            logger.info(f"Sync completed successfully for project: {project.name}")
+            logger.info(f"Background sync completed successfully for project: {project.name}")
 
             # Mark project as watching for changes after successful sync
             from basic_memory.services.sync_status_service import sync_status_tracker
 
             sync_status_tracker.start_project_watch(project.name)
             logger.info(f"Project {project.name} is now watching for changes")
         except Exception as e:  # pragma: no cover
-            logger.error(f"Error syncing project {project.name}: {e}")
+            logger.error(f"Error in background sync for project {project.name}: {e}")
             # Mark sync as failed for this project
             from basic_memory.services.sync_status_service import sync_status_tracker
 
             sync_status_tracker.fail_project_sync(project.name, str(e))
-            # Continue with other projects even if one fails
+
+    # Create background tasks for all project syncs (non-blocking)
+    sync_tasks = [
+        asyncio.create_task(sync_project_background(project)) for project in active_projects
+    ]
+    logger.info(f"Created {len(sync_tasks)} background sync tasks")
+
+    # Don't await the tasks - let them run in background while we continue
 
     # Then start the watch service in the background
     logger.info("Starting watch service for all projects")

src/basic_memory/sync/sync_service.py

@@ -1,7 +1,9 @@
 """Service for syncing files between filesystem and database."""
 
+import asyncio
 import os
 import time
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from datetime import datetime
 from pathlib import Path
@@ -80,6 +82,41 @@ def __init__(
         self.relation_repository = relation_repository
         self.search_service = search_service
         self.file_service = file_service
+        self._thread_pool = ThreadPoolExecutor(max_workers=app_config.sync_thread_pool_size)
+
+    async def _read_file_async(self, file_path: Path) -> str:
+        """Read file content in thread pool to avoid blocking the event loop."""
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(self._thread_pool, file_path.read_text, "utf-8")
+
+    async def _compute_checksum_async(self, path: str) -> str:
+        """Compute file checksum in thread pool to avoid blocking the event loop."""
+
+        def _sync_compute_checksum(path_str: str) -> str:
+            # Synchronous version for thread pool execution
+            path_obj = self.file_service.base_path / path_str
+
+            if self.file_service.is_markdown(path_str):
+                content = path_obj.read_text(encoding="utf-8")
+            else:
+                content = path_obj.read_bytes()
+
+            # Use the synchronous version of compute_checksum
+            import hashlib
+
+            if isinstance(content, str):
+                content_bytes = content.encode("utf-8")
+            else:
+                content_bytes = content
+            return hashlib.sha256(content_bytes).hexdigest()
+
+        loop = asyncio.get_event_loop()
+        return await loop.run_in_executor(self._thread_pool, _sync_compute_checksum, path)
+
+    def __del__(self):
+        """Cleanup thread pool when service is destroyed."""
+        if hasattr(self, "_thread_pool"):
+            self._thread_pool.shutdown(wait=False)
 
     async def sync(self, directory: Path, project_name: Optional[str] = None) -> SyncReport:
         """Sync all files with database."""
@@ -289,7 +326,7 @@ async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Optiona
         logger.debug(f"Parsing markdown file, path: {path}, new: {new}")
 
         file_path = self.entity_parser.base_path / path
-        file_content = file_path.read_text(encoding="utf-8")
+        file_content = await self._read_file_async(file_path)
         file_contains_frontmatter = has_frontmatter(file_content)
 
         # entity markdown will always contain front matter, so it can be used up create/update the entity
@@ -326,7 +363,7 @@ async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Optiona
         # After updating relations, we need to compute the checksum again
         # This is necessary for files with wikilinks to ensure consistent checksums
         # after relation processing is complete
-        final_checksum = await self.file_service.compute_checksum(path)
+        final_checksum = await self._compute_checksum_async(path)
 
         # set checksum
         await self.entity_repository.update(entity.id, {"checksum": final_checksum})
@@ -350,7 +387,7 @@ async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Optional
         Returns:
             Tuple of (entity, checksum)
         """
-        checksum = await self.file_service.compute_checksum(path)
+        checksum = await self._compute_checksum_async(path)
         if new:
             # Generate permalink from path
             await self.entity_service.resolve_permalink(path)
@@ -620,7 +657,7 @@ async def scan_directory(self, directory: Path) -> ScanResult:
 
                 path = Path(root) / filename
                 rel_path = path.relative_to(directory).as_posix()
-                checksum = await self.file_service.compute_checksum(rel_path)
+                checksum = await self._compute_checksum_async(rel_path)
                 result.files[rel_path] = checksum
                 result.checksums[checksum] = rel_path
 

tests/services/test_initialization.py

@@ -1,6 +1,5 @@
 """Tests for the initialization service."""
 
-from pathlib import Path
 from unittest.mock import patch, MagicMock, AsyncMock
 
 import pytest
@@ -127,10 +126,11 @@ async def test_reconcile_projects_with_error_handling(mock_get_db, app_config):
 @patch("basic_memory.services.initialization.db.get_or_create_db")
 @patch("basic_memory.cli.commands.sync.get_sync_service")
 @patch("basic_memory.sync.WatchService")
-async def test_initialize_file_sync_sequential(
-    mock_watch_service_class, mock_get_sync_service, mock_get_db, app_config
+@patch("basic_memory.services.initialization.asyncio.create_task")
+async def test_initialize_file_sync_background_tasks(
+    mock_create_task, mock_watch_service_class, mock_get_sync_service, mock_get_db, app_config
 ):
-    """Test file sync initialization with sequential project processing."""
+    """Test file sync initialization with background task processing."""
     # Setup mocks
     mock_session_maker = AsyncMock()
     mock_get_db.return_value = (None, mock_session_maker)
@@ -154,6 +154,11 @@ async def test_initialize_file_sync_sequential(
     mock_sync_service.sync = AsyncMock()
     mock_get_sync_service.return_value = mock_sync_service
 
+    # Mock background tasks
+    mock_task1 = MagicMock()
+    mock_task2 = MagicMock()
+    mock_create_task.side_effect = [mock_task1, mock_task2]
+
     # Mock the repository
     with patch("basic_memory.services.initialization.ProjectRepository") as mock_repo_class:
         mock_repo_class.return_value = mock_repository
@@ -165,22 +170,11 @@ async def test_initialize_file_sync_sequential(
         # Assertions
         mock_repository.get_active_projects.assert_called_once()
 
-        # Should call sync for each project sequentially
-        assert mock_get_sync_service.call_count == 2
-        mock_get_sync_service.assert_any_call(mock_project1)
-        mock_get_sync_service.assert_any_call(mock_project2)
+        # Should create background tasks for each project (non-blocking)
+        assert mock_create_task.call_count == 2
 
-        # Should call sync on each project
-        assert mock_sync_service.sync.call_count == 2
-        mock_sync_service.sync.assert_any_call(
-            Path(mock_project1.path), project_name=mock_project1.name
-        )
-        mock_sync_service.sync.assert_any_call(
-            Path(mock_project2.path), project_name=mock_project2.name
-        )
+        # Verify tasks were created but not awaited (function returns immediately)
+        assert result is None
 
-        # Should start the watch service
+        # Watch service should still be started
         mock_watch_service.run.assert_called_once()
-
-        # Should return None
-        assert result is None