admin/basicmachines-co-basic-memory
1
0
Fork 0
mirror of https://github.com/basicmachines-co/basic-memory synced 2026-06-21 13:47:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: Streaming Foundation & Async I/O Consolidation (SPEC-19) (#384)

Browse Source 
Signed-off-by: phernandez <paul@basicmachines.co>
Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Paul Hernandez
2025-10-21 09:03:59 -05:00
committed by GitHub
parent 32236cd247
commit e78345ff25
35 changed files with 3336 additions and 1595 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/basic_memory/alembic/versions/9d9c1cb7d8f5_add_mtime_and_size_columns_to_entity_.py
+49
View File
@@ -0,0 +1,49 @@
"""Add mtime and size columns to Entity for sync optimization
Revision ID: 9d9c1cb7d8f5
Revises: a1b2c3d4e5f6
Create Date: 2025-10-20 05:07:55.173849
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "9d9c1cb7d8f5"
down_revision: Union[str, None] = "a1b2c3d4e5f6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("entity", schema=None) as batch_op:
batch_op.add_column(sa.Column("mtime", sa.Float(), nullable=True))
batch_op.add_column(sa.Column("size", sa.Integer(), nullable=True))
batch_op.drop_constraint(batch_op.f("fk_entity_project_id"), type_="foreignkey")
batch_op.create_foreign_key(
batch_op.f("fk_entity_project_id"), "project", ["project_id"], ["id"]
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("entity", schema=None) as batch_op:
batch_op.drop_constraint(batch_op.f("fk_entity_project_id"), type_="foreignkey")
batch_op.create_foreign_key(
batch_op.f("fk_entity_project_id"),
"project",
["project_id"],
["id"],
ondelete="CASCADE",
)
batch_op.drop_column("size")
batch_op.drop_column("mtime")
# ### end Alembic commands ###
src/basic_memory/alembic/versions/e7e1f4367280_add_scan_watermark_tracking_to_project.py
+37
View File
@@ -0,0 +1,37 @@
"""Add scan watermark tracking to Project
Revision ID: e7e1f4367280
Revises: 9d9c1cb7d8f5
Create Date: 2025-10-20 16:42:46.625075
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "e7e1f4367280"
down_revision: Union[str, None] = "9d9c1cb7d8f5"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("project", schema=None) as batch_op:
batch_op.add_column(sa.Column("last_scan_timestamp", sa.Float(), nullable=True))
batch_op.add_column(sa.Column("last_file_count", sa.Integer(), nullable=True))
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
with op.batch_alter_table("project", schema=None) as batch_op:
batch_op.drop_column("last_file_count")
batch_op.drop_column("last_scan_timestamp")
# ### end Alembic commands ###
src/basic_memory/deps.py
+2
View File
@@ -321,6 +321,7 @@ async def get_sync_service(
entity_parser: EntityParserDep,
entity_repository: EntityRepositoryDep,
relation_repository: RelationRepositoryDep,
project_repository: ProjectRepositoryDep,
search_service: SearchServiceDep,
file_service: FileServiceDep,
) -> SyncService: # pragma: no cover
@@ -334,6 +335,7 @@ async def get_sync_service(
entity_parser=entity_parser,
entity_repository=entity_repository,
relation_repository=relation_repository,
project_repository=project_repository,
search_service=search_service,
file_service=file_service,
)
src/basic_memory/file_utils.py
+8 -83
View File
@@ -5,6 +5,7 @@ from pathlib import Path
import re
from typing import Any, Dict, Union
import aiofiles
import yaml
import frontmatter
from loguru import logger
@@ -52,29 +53,12 @@ async def compute_checksum(content: Union[str, bytes]) -> str:
raise FileError(f"Failed to compute checksum: {e}")
async def ensure_directory(path: FilePath) -> None:
"""
Ensure directory exists, creating if necessary.
Args:
path: Directory path to ensure (Path or string)
Raises:
FileWriteError: If directory creation fails
"""
try:
# Convert string to Path if needed
path_obj = Path(path) if isinstance(path, str) else path
path_obj.mkdir(parents=True, exist_ok=True)
except Exception as e: # pragma: no cover
logger.error("Failed to create directory", path=str(path), error=str(e))
raise FileWriteError(f"Failed to create directory {path}: {e}")
async def write_file_atomic(path: FilePath, content: str) -> None:
"""
Write file with atomic operation using temporary file.
Uses aiofiles for true async I/O (non-blocking).
Args:
path: Target file path (Path or string)
content: Content to write
@@ -87,7 +71,11 @@ async def write_file_atomic(path: FilePath, content: str) -> None:
temp_path = path_obj.with_suffix(".tmp")
try:
temp_path.write_text(content, encoding="utf-8")
# Use aiofiles for non-blocking write
async with aiofiles.open(temp_path, mode="w", encoding="utf-8") as f:
await f.write(content)
# Atomic rename (this is fast, doesn't need async)
temp_path.replace(path_obj)
logger.debug("Wrote file atomically", path=str(path_obj), content_length=len(content))
except Exception as e: # pragma: no cover
@@ -185,69 +173,6 @@ def remove_frontmatter(content: str) -> str:
return parts[2].strip()
async def update_frontmatter(path: FilePath, updates: Dict[str, Any]) -> str:
"""Update frontmatter fields in a file while preserving all content.
Only modifies the frontmatter section, leaving all content untouched.
Creates frontmatter section if none exists.
Returns checksum of updated file.
Args:
path: Path to markdown file (Path or string)
updates: Dict of frontmatter fields to update
Returns:
Checksum of updated file
Raises:
FileError: If file operations fail
ParseError: If frontmatter parsing fails
"""
try:
# Convert string to Path if needed
path_obj = Path(path) if isinstance(path, str) else path
# Read current content
content = path_obj.read_text(encoding="utf-8")
# Parse current frontmatter with proper error handling for malformed YAML
current_fm = {}
if has_frontmatter(content):
try:
current_fm = parse_frontmatter(content)
content = remove_frontmatter(content)
except (ParseError, yaml.YAMLError) as e:
# Log warning and treat as plain markdown without frontmatter
logger.warning(
f"Failed to parse YAML frontmatter in {path_obj}: {e}. "
"Treating file as plain markdown without frontmatter."
)
# Keep full content, treat as having no frontmatter
current_fm = {}
# Update frontmatter
new_fm = {**current_fm, **updates}
# Write new file with updated frontmatter
yaml_fm = yaml.dump(new_fm, sort_keys=False, allow_unicode=True)
final_content = f"---\n{yaml_fm}---\n\n{content.strip()}"
logger.debug("Updating frontmatter", path=str(path_obj), update_keys=list(updates.keys()))
await write_file_atomic(path_obj, final_content)
return await compute_checksum(final_content)
except Exception as e: # pragma: no cover
# Only log real errors (not YAML parsing, which is handled above)
if not isinstance(e, (ParseError, yaml.YAMLError)):
logger.error(
"Failed to update frontmatter",
path=str(path) if isinstance(path, (str, Path)) else "<unknown>",
error=str(e),
)
raise FileError(f"Failed to update frontmatter: {e}")
def dump_frontmatter(post: frontmatter.Post) -> str:
"""
Serialize frontmatter.Post to markdown with Obsidian-compatible YAML format.
src/basic_memory/mcp/tools/__init__.py
-2
View File
@@ -18,7 +18,6 @@ from basic_memory.mcp.tools.canvas import canvas
from basic_memory.mcp.tools.list_directory import list_directory
from basic_memory.mcp.tools.edit_note import edit_note
from basic_memory.mcp.tools.move_note import move_note
from basic_memory.mcp.tools.sync_status import sync_status
from basic_memory.mcp.tools.project_management import (
list_memory_projects,
create_memory_project,
@@ -44,7 +43,6 @@ __all__ = [
"recent_activity",
"search",
"search_notes",
"sync_status",
"view_note",
"write_note",
]
src/basic_memory/mcp/tools/build_context.py
-22
View File
@@ -106,28 +106,6 @@ async def build_context(
# Get the active project using the new stateless approach
active_project = await get_active_project(client, project, context)
# Check migration status and wait briefly if needed
from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
migration_status = await wait_for_migration_or_return_status(
timeout=5.0, project_name=active_project.name
)
if migration_status: # pragma: no cover
# Return a proper GraphContext with status message
from basic_memory.schemas.memory import MemoryMetadata
from datetime import datetime
return GraphContext(
results=[],
metadata=MemoryMetadata(
depth=depth or 1,
timeframe=timeframe,
generated_at=datetime.now().astimezone(),
primary_count=0,
related_count=0,
uri=migration_status, # Include status in metadata
),
)
project_url = active_project.project_url
response = await call_get(
src/basic_memory/mcp/tools/read_note.py
-8
View File
@@ -97,14 +97,6 @@ async def read_note(
)
return f"# Error\n\nIdentifier '{identifier}' is not allowed - paths must stay within project boundaries"
# Check migration status and wait briefly if needed
from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
migration_status = await wait_for_migration_or_return_status(
timeout=5.0, project_name=active_project.name
)
if migration_status: # pragma: no cover
return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before reading notes."
project_url = active_project.project_url
# Get the file via REST API - first try direct permalink lookup
src/basic_memory/mcp/tools/sync_status.py
-261
View File
@@ -1,261 +0,0 @@
"""Sync status tool for Basic Memory MCP server."""
from typing import Optional
from loguru import logger
from fastmcp import Context
from basic_memory.config import ConfigManager
from basic_memory.mcp.async_client import get_client
from basic_memory.mcp.server import mcp
from basic_memory.mcp.project_context import get_active_project
from basic_memory.services.sync_status_service import sync_status_tracker
def _get_all_projects_status() -> list[str]:
"""Get status lines for all configured projects."""
status_lines = []
try:
app_config = ConfigManager().config
if app_config.projects:
status_lines.extend(["", "---", "", "**All Projects Status:**"])
for project_name, project_path in app_config.projects.items():
# Check if this project has sync status
project_sync_status = sync_status_tracker.get_project_status(project_name)
if project_sync_status:
# Project has tracked sync activity
if project_sync_status.status.value == "watching":
# Project is actively watching for changes (steady state)
status_icon = "👁️"
status_text = "Watching for changes"
elif project_sync_status.status.value == "completed":
# Sync completed but not yet watching - transitional state
status_icon = "✅"
status_text = "Sync completed"
elif project_sync_status.status.value in ["scanning", "syncing"]:
status_icon = "🔄"
status_text = "Sync in progress"
if project_sync_status.files_total > 0:
progress_pct = (
project_sync_status.files_processed
/ project_sync_status.files_total
) * 100
status_text += f" ({project_sync_status.files_processed}/{project_sync_status.files_total}, {progress_pct:.0f}%)"
elif project_sync_status.status.value == "failed":
status_icon = "❌"
status_text = f"Sync error: {project_sync_status.error or 'Unknown error'}"
else:
status_icon = "⏸️"
status_text = project_sync_status.status.value.title()
else:
# Project has no tracked sync activity - will be synced automatically
status_icon = "⏳"
status_text = "Pending sync"
status_lines.append(f"- {status_icon} **{project_name}**: {status_text}")
except Exception as e:
logger.debug(f"Could not get project config for comprehensive status: {e}")
return status_lines
@mcp.tool(
description="""Check the status of file synchronization and background operations.
Use this tool to:
- Check if file sync is in progress or completed
- Get detailed sync progress information
- Understand if your files are fully indexed
- Get specific error details if sync operations failed
- Monitor initial project setup and legacy migration
This covers all sync operations including:
- Initial project setup and file indexing
- Legacy project migration to unified database
- Ongoing file monitoring and updates
- Background processing of knowledge graphs
""",
)
async def sync_status(project: Optional[str] = None, context: Context | None = None) -> str:
"""Get current sync status and system readiness information.
This tool provides detailed information about any ongoing or completed
sync operations, helping users understand when their files are ready.
Args:
project: Optional project name to get project-specific context
Returns:
Formatted sync status with progress, readiness, and guidance
"""
logger.info("MCP tool call tool=sync_status")
async with get_client() as client:
status_lines = []
try:
from basic_memory.services.sync_status_service import sync_status_tracker
# Get overall summary
summary = sync_status_tracker.get_summary()
is_ready = sync_status_tracker.is_ready
# Header
status_lines.extend(
[
"# Basic Memory Sync Status",
"",
f"**Current Status**: {summary}",
f"**System Ready**: {'✅ Yes' if is_ready else '🔄 Processing'}",
"",
]
)
if is_ready:
status_lines.extend(
[
"✅ **All sync operations completed**",
"",
"- File indexing is complete",
"- Knowledge graphs are up to date",
"- All Basic Memory tools are fully operational",
"",
"Your knowledge base is ready for use!",
]
)
# Show all projects status even when ready
status_lines.extend(_get_all_projects_status())
else:
# System is still processing - show both active and all projects
all_sync_projects = sync_status_tracker.get_all_projects()
active_projects = [
p
for p in all_sync_projects.values()
if p.status.value in ["scanning", "syncing"]
]
failed_projects = [
p for p in all_sync_projects.values() if p.status.value == "failed"
]
if active_projects:
status_lines.extend(
[
"🔄 **File synchronization in progress**",
"",
"Basic Memory is automatically processing all configured projects and building knowledge graphs.",
"This typically takes 1-3 minutes depending on the amount of content.",
"",
"**Currently Processing:**",
]
)
for project_status in active_projects:
progress = ""
if project_status.files_total > 0:
progress_pct = (
project_status.files_processed / project_status.files_total
) * 100
progress = f" ({project_status.files_processed}/{project_status.files_total}, {progress_pct:.0f}%)"
status_lines.append(
f"- **{project_status.project_name}**: {project_status.message}{progress}"
)
status_lines.extend(
[
"",
"**What's happening:**",
"- Scanning and indexing markdown files",
"- Building entity and relationship graphs",
"- Settings up full-text search indexes",
"- Processing file changes and updates",
"",
"**What you can do:**",
"- Wait for automatic processing to complete - no action needed",
"- Use this tool again to check progress",
"- Simple operations may work already",
"- All projects will be available once sync finishes",
]
)
# Handle failed projects (independent of active projects)
if failed_projects:
status_lines.extend(["", "❌ **Some projects failed to sync:**", ""])
for project_status in failed_projects:
status_lines.append(
f"- **{project_status.project_name}**: {project_status.error or 'Unknown error'}"
)
status_lines.extend(
[
"",
"**Next steps:**",
"1. Check the logs for detailed error information",
"2. Ensure file permissions allow read/write access",
"3. Try restarting the MCP server",
"4. If issues persist, consider filing a support issue",
]
)
elif not active_projects:
# No active or failed projects - must be pending
status_lines.extend(
[
"⏳ **Sync operations pending**",
"",
"File synchronization has been queued but hasn't started yet.",
"This usually resolves automatically within a few seconds.",
]
)
# Add comprehensive project status for all configured projects
all_projects_status = _get_all_projects_status()
if all_projects_status:
status_lines.extend(all_projects_status)
# Add explanation about automatic syncing if there are unsynced projects
unsynced_count = sum(1 for line in all_projects_status if "⏳" in line)
if unsynced_count > 0 and not is_ready:
status_lines.extend(
[
"",
"**Note**: All configured projects will be automatically synced during startup.",
]
)
# Add project context if provided
if project:
try:
active_project = await get_active_project(client, project, context)
status_lines.extend(
[
"",
"---",
"",
f"**Active Project**: {active_project.name}",
f"**Project Path**: {active_project.home}",
]
)
except Exception as e:
logger.debug(f"Could not get project info: {e}")
return "\n".join(status_lines)
except Exception as e:
return f"""# Sync Status - Error
❌ **Unable to check sync status**: {str(e)}
**Troubleshooting:**
- The system may still be starting up
- Try waiting a few seconds and checking again
- Check logs for detailed error information
- Consider restarting if the issue persists
"""
src/basic_memory/mcp/tools/utils.py
-70
View File
@@ -510,73 +510,3 @@ async def call_delete(
except HTTPStatusError as e:
raise ToolError(error_message) from e
def check_migration_status() -> Optional[str]:
"""Check if sync/migration is in progress and return status message if so.
Returns:
Status message if sync is in progress, None if system is ready
"""
try:
from basic_memory.services.sync_status_service import sync_status_tracker
if not sync_status_tracker.is_ready:
return sync_status_tracker.get_summary()
return None
except Exception:
# If there's any error checking sync status, assume ready
return None
async def wait_for_migration_or_return_status(
timeout: float = 5.0, project_name: Optional[str] = None
) -> Optional[str]:
"""Wait briefly for sync/migration to complete, or return status message.
Args:
timeout: Maximum time to wait for sync completion
project_name: Optional project name to check specific project status.
If provided, only checks that project's readiness.
If None, uses global status check (legacy behavior).
Returns:
Status message if sync is still in progress, None if ready
"""
try:
from basic_memory.services.sync_status_service import sync_status_tracker
import asyncio
# Check if we should use project-specific or global status
def is_ready() -> bool:
if project_name:
return sync_status_tracker.is_project_ready(project_name)
return sync_status_tracker.is_ready
if is_ready():
return None
# Wait briefly for sync to complete
start_time = asyncio.get_event_loop().time()
while (asyncio.get_event_loop().time() - start_time) < timeout:
if is_ready():
return None
await asyncio.sleep(0.1) # Check every 100ms
# Still not ready after timeout
if project_name:
# For project-specific checks, get project status details
project_status = sync_status_tracker.get_project_status(project_name)
if project_status and project_status.status.value == "failed":
error_msg = project_status.error or "Unknown sync error"
return f"❌ Sync failed for project '{project_name}': {error_msg}"
elif project_status:
return f"🔄 Project '{project_name}' is still syncing: {project_status.message}"
else:
return f"⚠️ Project '{project_name}' status unknown"
else:
# Fall back to global summary for legacy calls
return sync_status_tracker.get_summary()
except Exception: # pragma: no cover
# If there's any error, assume ready
return None
src/basic_memory/mcp/tools/write_note.py
-9
View File
@@ -140,15 +140,6 @@ async def write_note(
)
return f"# Error\n\nFolder path '{folder}' is not allowed - paths must stay within project boundaries"
# Check migration status and wait briefly if needed
from basic_memory.mcp.tools.utils import wait_for_migration_or_return_status
migration_status = await wait_for_migration_or_return_status(
timeout=5.0, project_name=active_project.name
)
if migration_status: # pragma: no cover
return f"# System Status\n\n{migration_status}\n\nPlease wait for migration to complete before creating notes."
# Process tags using the helper function
tag_list = parse_tags(tags)
# Create the entity request
src/basic_memory/models/knowledge.py
+7
View File
@@ -13,6 +13,7 @@ from sqlalchemy import (
DateTime,
Index,
JSON,
Float,
text,
)
from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -73,6 +74,12 @@ class Entity(Base):
# checksum of file
checksum: Mapped[Optional[str]] = mapped_column(String, nullable=True)
# File metadata for sync
# mtime: file modification timestamp (Unix epoch float) for change detection
mtime: Mapped[Optional[float]] = mapped_column(Float, nullable=True)
# size: file size in bytes for quick change detection
size: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
# Metadata and tracking
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=lambda: datetime.now().astimezone()
src/basic_memory/models/project.py
+5
View File
@@ -9,6 +9,7 @@ from sqlalchemy import (
Text,
Boolean,
DateTime,
Float,
Index,
event,
)
@@ -61,6 +62,10 @@ class Project(Base):
onupdate=lambda: datetime.now(UTC),
)
# Sync optimization - scan watermark tracking
last_scan_timestamp: Mapped[Optional[float]] = mapped_column(Float, nullable=True)
last_file_count: Mapped[Optional[int]] = mapped_column(Integer, nullable=True)
# Define relationships to entities, observations, and relations
# These relationships will be established once we add project_id to those models
entities = relationship("Entity", back_populates="project", cascade="all, delete-orphan")
src/basic_memory/repository/entity_repository.py
+32
View File
@@ -63,6 +63,23 @@ class EntityRepository(Repository[Entity]):
)
return await self.find_one(query)
async def find_by_checksum(self, checksum: str) -> Sequence[Entity]:
"""Find entities with the given checksum.
Used for move detection - finds entities that may have been moved to a new path.
Multiple entities may have the same checksum if files were copied.
Args:
checksum: File content checksum to search for
Returns:
Sequence of entities with matching checksum (may be empty)
"""
query = self.select().where(Entity.checksum == checksum)
# Don't load relationships for move detection - we only need file_path and checksum
result = await self.execute_query(query, use_query_options=False)
return list(result.scalars().all())
async def delete_by_file_path(self, file_path: Union[Path, str]) -> bool:
"""Delete entity with the provided file_path.
@@ -197,6 +214,21 @@ class EntityRepository(Repository[Entity]):
entity = await self._handle_permalink_conflict(entity, session)
return entity
async def get_all_file_paths(self) -> List[str]:
"""Get all file paths for this project - optimized for deletion detection.
Returns only file_path strings without loading entities or relationships.
Used by streaming sync to detect deleted files efficiently.
Returns:
List of file_path strings for all entities in the project
"""
query = select(Entity.file_path)
query = self._add_project_filter(query)
result = await self.execute_query(query, use_query_options=False)
return list(result.scalars().all())
async def get_distinct_directories(self) -> List[str]:
"""Extract unique directory paths from file_path column.
src/basic_memory/services/file_service.py
+162 -24
View File
@@ -1,12 +1,17 @@
"""Service for file operations with checksum tracking."""
import asyncio
import hashlib
import mimetypes
from os import stat_result
from pathlib import Path
from typing import Any, Dict, Tuple, Union
import aiofiles
import yaml
from basic_memory import file_utils
from basic_memory.file_utils import FileError
from basic_memory.file_utils import FileError, ParseError
from basic_memory.markdown.markdown_processor import MarkdownProcessor
from basic_memory.models import Entity as EntityModel
from basic_memory.schemas import Entity as EntitySchema
@@ -16,13 +21,15 @@ from loguru import logger
class FileService:
"""Service for handling file operations.
"""Service for handling file operations with concurrency control.
All paths are handled as Path objects internally. Strings are converted to
Path objects when passed in. Relative paths are assumed to be relative to
base_path.
Features:
- True async I/O with aiofiles (non-blocking)
- Built-in concurrency limits (semaphore)
- Consistent file writing with checksums
- Frontmatter management
- Atomic operations
@@ -33,9 +40,13 @@ class FileService:
self,
base_path: Path,
markdown_processor: MarkdownProcessor,
max_concurrent_files: int = 10,
):
self.base_path = base_path.resolve() # Get absolute path
self.markdown_processor = markdown_processor
# Semaphore to limit concurrent file operations
# Prevents OOM on large projects by processing files in batches
self._file_semaphore = asyncio.Semaphore(max_concurrent_files)
def get_entity_path(self, entity: Union[EntityModel, EntitySchema]) -> Path:
"""Generate absolute filesystem path for entity.
@@ -104,6 +115,33 @@ class FileService:
logger.error("Failed to check file existence", path=str(path), error=str(e))
raise FileOperationError(f"Failed to check file existence: {e}")
async def ensure_directory(self, path: FilePath) -> None:
"""Ensure directory exists, creating if necessary.
Uses semaphore to control concurrency for directory creation operations.
Args:
path: Directory path to ensure (Path or string)
Raises:
FileOperationError: If directory creation fails
"""
try:
# Convert string to Path if needed
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
# Use semaphore for concurrency control
async with self._file_semaphore:
# Run blocking mkdir in thread pool
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None, lambda: full_path.mkdir(parents=True, exist_ok=True)
)
except Exception as e: # pragma: no cover
logger.error("Failed to create directory", path=str(path), error=str(e))
raise FileOperationError(f"Failed to create directory {path}: {e}")
async def write_file(self, path: FilePath, content: str) -> str:
"""Write content to file and return checksum.
@@ -126,7 +164,7 @@ class FileService:
try:
# Ensure parent directory exists
await file_utils.ensure_directory(full_path.parent)
await self.ensure_directory(full_path.parent)
# Write content atomically
logger.info(
@@ -147,9 +185,45 @@ class FileService:
logger.exception("File write error", path=str(full_path), error=str(e))
raise FileOperationError(f"Failed to write file: {e}")
# TODO remove read_file
async def read_file_content(self, path: FilePath) -> str:
"""Read file content using true async I/O with aiofiles.
Handles both absolute and relative paths. Relative paths are resolved
against base_path.
Args:
path: Path to read (Path or string)
Returns:
File content as string
Raises:
FileOperationError: If read fails
"""
# Convert string to Path if needed
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
try:
logger.debug("Reading file content", operation="read_file_content", path=str(full_path))
async with aiofiles.open(full_path, mode="r", encoding="utf-8") as f:
content = await f.read()
logger.debug(
"File read completed",
path=str(full_path),
content_length=len(content),
)
return content
except Exception as e:
logger.exception("File read error", path=str(full_path), error=str(e))
raise FileOperationError(f"Failed to read file: {e}")
async def read_file(self, path: FilePath) -> Tuple[str, str]:
"""Read file and compute checksum.
"""Read file and compute checksum using true async I/O.
Uses aiofiles for non-blocking file reads.
Handles both absolute and relative paths. Relative paths are resolved
against base_path.
@@ -169,7 +243,11 @@ class FileService:
try:
logger.debug("Reading file", operation="read_file", path=str(full_path))
content = full_path.read_text(encoding="utf-8")
# Use aiofiles for non-blocking read
async with aiofiles.open(full_path, mode="r", encoding="utf-8") as f:
content = await f.read()
checksum = await file_utils.compute_checksum(content)
logger.debug(
@@ -199,29 +277,85 @@ class FileService:
full_path.unlink(missing_ok=True)
async def update_frontmatter(self, path: FilePath, updates: Dict[str, Any]) -> str:
"""
Update frontmatter fields in a file while preserving all content.
"""Update frontmatter fields in a file while preserving all content.
Only modifies the frontmatter section, leaving all content untouched.
Creates frontmatter section if none exists.
Returns checksum of updated file.
Uses aiofiles for true async I/O (non-blocking).
Args:
path: Path to the file (Path or string)
updates: Dictionary of frontmatter fields to update
path: Path to markdown file (Path or string)
updates: Dict of frontmatter fields to update
Returns:
Checksum of updated file
Raises:
FileOperationError: If file operations fail
ParseError: If frontmatter parsing fails
"""
# Convert string to Path if needed
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
return await file_utils.update_frontmatter(full_path, updates)
try:
# Read current content using aiofiles
async with aiofiles.open(full_path, mode="r", encoding="utf-8") as f:
content = await f.read()
# Parse current frontmatter with proper error handling for malformed YAML
current_fm = {}
if file_utils.has_frontmatter(content):
try:
current_fm = file_utils.parse_frontmatter(content)
content = file_utils.remove_frontmatter(content)
except (ParseError, yaml.YAMLError) as e:
# Log warning and treat as plain markdown without frontmatter
logger.warning(
f"Failed to parse YAML frontmatter in {full_path}: {e}. "
"Treating file as plain markdown without frontmatter."
)
# Keep full content, treat as having no frontmatter
current_fm = {}
# Update frontmatter
new_fm = {**current_fm, **updates}
# Write new file with updated frontmatter
yaml_fm = yaml.dump(new_fm, sort_keys=False, allow_unicode=True)
final_content = f"---\n{yaml_fm}---\n\n{content.strip()}"
logger.debug(
"Updating frontmatter", path=str(full_path), update_keys=list(updates.keys())
)
await file_utils.write_file_atomic(full_path, final_content)
return await file_utils.compute_checksum(final_content)
except Exception as e:
# Only log real errors (not YAML parsing, which is handled above)
if not isinstance(e, (ParseError, yaml.YAMLError)):
logger.error(
"Failed to update frontmatter",
path=str(full_path),
error=str(e),
)
raise FileOperationError(f"Failed to update frontmatter: {e}")
async def compute_checksum(self, path: FilePath) -> str:
"""Compute checksum for a file.
"""Compute checksum for a file using true async I/O.
Uses aiofiles for non-blocking I/O with 64KB chunked reading.
Semaphore limits concurrent file operations to prevent OOM.
Memory usage is constant regardless of file size.
Args:
path: Path to the file (Path or string)
Returns:
Checksum of the file content
SHA256 checksum hex string
Raises:
FileError: If checksum computation fails
@@ -230,18 +364,22 @@ class FileService:
path_obj = self.base_path / path if isinstance(path, str) else path
full_path = path_obj if path_obj.is_absolute() else self.base_path / path_obj
try:
if self.is_markdown(path):
# read str
content = full_path.read_text(encoding="utf-8")
else:
# read bytes
content = full_path.read_bytes()
return await file_utils.compute_checksum(content)
# Semaphore controls concurrency - max N files processed at once
async with self._file_semaphore:
try:
hasher = hashlib.sha256()
chunk_size = 65536 # 64KB chunks
except Exception as e: # pragma: no cover
logger.error("Failed to compute checksum", path=str(full_path), error=str(e))
raise FileError(f"Failed to compute checksum for {path}: {e}")
# async I/O with aiofiles
async with aiofiles.open(full_path, mode="rb") as f:
while chunk := await f.read(chunk_size):
hasher.update(chunk)
return hasher.hexdigest()
except Exception as e: # pragma: no cover
logger.error("Failed to compute checksum", path=str(full_path), error=str(e))
raise FileError(f"Failed to compute checksum for {path}: {e}")
def file_stats(self, path: FilePath) -> stat_result:
"""Return file stats for a given path.
src/basic_memory/services/initialization.py
-10
View File
@@ -118,18 +118,8 @@ async def initialize_file_sync(
sync_dir = Path(project.path)
await sync_service.sync(sync_dir, project_name=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 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))
# Create background tasks for all project syncs (non-blocking)
sync_tasks = [
src/basic_memory/services/sync_status_service.py
-198
View File
@@ -1,198 +0,0 @@
"""Simple sync status tracking service."""
from dataclasses import dataclass
from enum import Enum
from typing import Dict, Optional
class SyncStatus(Enum):
"""Status of sync operations."""
IDLE = "idle"
SCANNING = "scanning"
SYNCING = "syncing"
COMPLETED = "completed"
FAILED = "failed"
WATCHING = "watching"
@dataclass
class ProjectSyncStatus:
"""Sync status for a single project."""
project_name: str
status: SyncStatus
message: str = ""
files_total: int = 0
files_processed: int = 0
error: Optional[str] = None
class SyncStatusTracker:
"""Global tracker for all sync operations."""
def __init__(self):
self._project_statuses: Dict[str, ProjectSyncStatus] = {}
self._global_status: SyncStatus = SyncStatus.IDLE
def start_project_sync(self, project_name: str, files_total: int = 0) -> None:
"""Start tracking sync for a project."""
self._project_statuses[project_name] = ProjectSyncStatus(
project_name=project_name,
status=SyncStatus.SCANNING,
message="Scanning files",
files_total=files_total,
files_processed=0,
)
self._update_global_status()
def update_project_progress( # pragma: no cover
self,
project_name: str,
status: SyncStatus,
message: str = "",
files_processed: int = 0,
files_total: Optional[int] = None,
) -> None:
"""Update progress for a project."""
if project_name not in self._project_statuses: # pragma: no cover
return
project_status = self._project_statuses[project_name]
project_status.status = status
project_status.message = message
project_status.files_processed = files_processed
if files_total is not None:
project_status.files_total = files_total
self._update_global_status()
def complete_project_sync(self, project_name: str) -> None:
"""Mark project sync as completed."""
if project_name in self._project_statuses:
self._project_statuses[project_name].status = SyncStatus.COMPLETED
self._project_statuses[project_name].message = "Sync completed"
self._update_global_status()
def fail_project_sync(self, project_name: str, error: str) -> None:
"""Mark project sync as failed."""
if project_name in self._project_statuses:
self._project_statuses[project_name].status = SyncStatus.FAILED
self._project_statuses[project_name].error = error
self._update_global_status()
def start_project_watch(self, project_name: str) -> None:
"""Mark project as watching for changes (steady state after sync)."""
if project_name in self._project_statuses:
self._project_statuses[project_name].status = SyncStatus.WATCHING
self._project_statuses[project_name].message = "Watching for changes"
self._update_global_status()
else:
# Create new status if project isn't tracked yet
self._project_statuses[project_name] = ProjectSyncStatus(
project_name=project_name,
status=SyncStatus.WATCHING,
message="Watching for changes",
files_total=0,
files_processed=0,
)
self._update_global_status()
def _update_global_status(self) -> None:
"""Update global status based on project statuses."""
if not self._project_statuses: # pragma: no cover
self._global_status = SyncStatus.IDLE
return
statuses = [p.status for p in self._project_statuses.values()]
if any(s == SyncStatus.FAILED for s in statuses):
self._global_status = SyncStatus.FAILED
elif any(s in (SyncStatus.SCANNING, SyncStatus.SYNCING) for s in statuses):
self._global_status = SyncStatus.SYNCING
elif all(s in (SyncStatus.COMPLETED, SyncStatus.WATCHING) for s in statuses):
self._global_status = SyncStatus.COMPLETED
else:
self._global_status = SyncStatus.SYNCING
@property
def global_status(self) -> SyncStatus:
"""Get overall sync status."""
return self._global_status
@property
def is_syncing(self) -> bool:
"""Check if any sync operation is in progress."""
return self._global_status in (SyncStatus.SCANNING, SyncStatus.SYNCING)
@property
def is_ready(self) -> bool: # pragma: no cover
"""Check if system is ready (no sync in progress)."""
return self._global_status in (SyncStatus.IDLE, SyncStatus.COMPLETED)
def is_project_ready(self, project_name: str) -> bool:
"""Check if a specific project is ready for operations.
Args:
project_name: Name of the project to check
Returns:
True if the project is ready (completed, watching, or not tracked),
False if the project is syncing, scanning, or failed
"""
project_status = self._project_statuses.get(project_name)
if not project_status:
# Project not tracked = ready (likely hasn't been synced yet)
return True
return project_status.status in (SyncStatus.COMPLETED, SyncStatus.WATCHING, SyncStatus.IDLE)
def get_project_status(self, project_name: str) -> Optional[ProjectSyncStatus]:
"""Get status for a specific project."""
return self._project_statuses.get(project_name)
def get_all_projects(self) -> Dict[str, ProjectSyncStatus]:
"""Get all project statuses."""
return self._project_statuses.copy()
def get_summary(self) -> str: # pragma: no cover
"""Get a user-friendly summary of sync status."""
if self._global_status == SyncStatus.IDLE:
return "✅ System ready"
elif self._global_status == SyncStatus.COMPLETED:
return "✅ All projects synced successfully"
elif self._global_status == SyncStatus.FAILED:
failed_projects = [
p.project_name
for p in self._project_statuses.values()
if p.status == SyncStatus.FAILED
]
return f"❌ Sync failed for: {', '.join(failed_projects)}"
else:
active_projects = [
p.project_name
for p in self._project_statuses.values()
if p.status in (SyncStatus.SCANNING, SyncStatus.SYNCING)
]
total_files = sum(p.files_total for p in self._project_statuses.values())
processed_files = sum(p.files_processed for p in self._project_statuses.values())
if total_files > 0:
progress_pct = (processed_files / total_files) * 100
return f"🔄 Syncing {len(active_projects)} projects ({processed_files}/{total_files} files, {progress_pct:.0f}%)"
else:
return f"🔄 Syncing {len(active_projects)} projects"
def clear_completed(self) -> None:
"""Remove completed project statuses to clean up memory."""
self._project_statuses = {
name: status
for name, status in self._project_statuses.items()
if status.status != SyncStatus.COMPLETED
}
self._update_global_status()
# Global sync status tracker instance
sync_status_tracker = SyncStatusTracker()
src/basic_memory/sync/sync_service.py
+460 -248
View File
@@ -4,14 +4,14 @@ import asyncio
import os
import time
from collections import OrderedDict
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Dict, List, Optional, Set, Tuple
from typing import AsyncIterator, Dict, List, Optional, Set, Tuple
import aiofiles.os
import logfire
from loguru import logger
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from basic_memory import db
@@ -20,13 +20,17 @@ from basic_memory.file_utils import has_frontmatter
from basic_memory.ignore_utils import load_bmignore_patterns, should_ignore_path
from basic_memory.markdown import EntityParser, MarkdownProcessor
from basic_memory.models import Entity, Project
from basic_memory.repository import EntityRepository, RelationRepository, ObservationRepository
from basic_memory.repository import (
EntityRepository,
RelationRepository,
ObservationRepository,
ProjectRepository,
)
from basic_memory.repository.search_repository import SearchRepository
from basic_memory.services import EntityService, FileService
from basic_memory.services.exceptions import SyncFatalError
from basic_memory.services.link_resolver import LinkResolver
from basic_memory.services.search_service import SearchService
from basic_memory.services.sync_status_service import sync_status_tracker, SyncStatus
# Circuit breaker configuration
MAX_CONSECUTIVE_FAILURES = 3
@@ -120,6 +124,7 @@ class SyncService:
entity_parser: EntityParser,
entity_repository: EntityRepository,
relation_repository: RelationRepository,
project_repository: ProjectRepository,
search_service: SearchService,
file_service: FileService,
):
@@ -128,60 +133,15 @@ class SyncService:
self.entity_parser = entity_parser
self.entity_repository = entity_repository
self.relation_repository = relation_repository
self.project_repository = project_repository
self.search_service = search_service
self.file_service = file_service
self._thread_pool = ThreadPoolExecutor(max_workers=app_config.sync_thread_pool_size)
# Load ignore patterns once at initialization for performance
self._ignore_patterns = load_bmignore_patterns()
# Circuit breaker: track file failures to prevent infinite retry loops
# Use OrderedDict for LRU behavior with bounded size to prevent unbounded memory growth
self._file_failures: OrderedDict[str, FileFailureInfo] = OrderedDict()
self._max_tracked_failures = 100 # Limit failure cache size
# Semaphore to limit concurrent file operations and prevent OOM on large projects
# Limits peak memory usage by processing files in batches rather than all at once
self._file_semaphore = asyncio.Semaphore(app_config.sync_max_concurrent_files)
async def _read_file_async(self, file_path: Path) -> str:
"""Read file content in thread pool to avoid blocking the event loop.
Uses semaphore to limit concurrent file reads and prevent OOM on large projects.
"""
async with self._file_semaphore:
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.
Uses semaphore to limit concurrent file reads and prevent OOM on large projects.
"""
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()
async with self._file_semaphore:
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 _should_skip_file(self, path: str) -> bool:
"""Check if file should be skipped due to repeated failures.
@@ -206,7 +166,7 @@ class SyncService:
# Compute current checksum to see if file changed
try:
current_checksum = await self._compute_checksum_async(path)
current_checksum = await self.file_service.compute_checksum(path)
# If checksum changed, file was modified - reset and retry
if current_checksum != failure_info.last_checksum:
@@ -236,7 +196,7 @@ class SyncService:
# Compute checksum for failure tracking
try:
checksum = await self._compute_checksum_async(path)
checksum = await self.file_service.compute_checksum(path)
except Exception:
# If checksum fails, use empty string (better than crashing)
checksum = ""
@@ -255,12 +215,17 @@ class SyncService:
f"path={path}, error={error}"
)
# Record metric for file failure
logfire.metric_counter("sync.circuit_breaker.failures").add(1)
# Log when threshold is reached
if failure_info.count >= MAX_CONSECUTIVE_FAILURES:
logger.error(
f"File {path} has failed {MAX_CONSECUTIVE_FAILURES} times and will be skipped. "
f"First failure: {failure_info.first_failure}, Last error: {error}"
)
# Record metric for file being blocked by circuit breaker
logfire.metric_counter("sync.circuit_breaker.blocked_files").add(1)
else:
# Create new failure record
self._file_failures[path] = FileFailureInfo(
@@ -290,128 +255,113 @@ class SyncService:
logger.info(f"Clearing failure history for {path} after successful sync")
del self._file_failures[path]
@logfire.instrument()
async def sync(self, directory: Path, project_name: Optional[str] = None) -> SyncReport:
"""Sync all files with database."""
"""Sync all files with database and update scan watermark."""
start_time = time.time()
sync_start_timestamp = time.time() # Capture at start for watermark
logger.info(f"Sync operation started for directory: {directory}")
# Start tracking sync for this project if project name provided
if project_name:
sync_status_tracker.start_project_sync(project_name)
# initial paths from db to sync
# path -> checksum
report = await self.scan(directory)
# Update progress with file counts
if project_name:
sync_status_tracker.update_project_progress(
project_name=project_name,
status=SyncStatus.SYNCING,
message="Processing file changes",
files_total=report.total,
files_processed=0,
)
# order of sync matters to resolve relations effectively
logger.info(
f"Sync changes detected: new_files={len(report.new)}, modified_files={len(report.modified)}, "
+ f"deleted_files={len(report.deleted)}, moved_files={len(report.moves)}"
)
files_processed = 0
# sync moves first
for old_path, new_path in report.moves.items():
# in the case where a file has been deleted and replaced by another file
# it will show up in the move and modified lists, so handle it in modified
if new_path in report.modified:
report.modified.remove(new_path)
logger.debug(
f"File marked as moved and modified: old_path={old_path}, new_path={new_path}"
)
else:
await self.handle_move(old_path, new_path)
files_processed += 1
if project_name:
sync_status_tracker.update_project_progress( # pragma: no cover
project_name=project_name,
status=SyncStatus.SYNCING,
message="Processing moves",
files_processed=files_processed,
)
with logfire.span("process_moves", move_count=len(report.moves)):
for old_path, new_path in report.moves.items():
# in the case where a file has been deleted and replaced by another file
# it will show up in the move and modified lists, so handle it in modified
if new_path in report.modified:
report.modified.remove(new_path)
logger.debug(
f"File marked as moved and modified: old_path={old_path}, new_path={new_path}"
)
else:
await self.handle_move(old_path, new_path)
# deleted next
for path in report.deleted:
await self.handle_delete(path)
files_processed += 1
if project_name:
sync_status_tracker.update_project_progress( # pragma: no cover
project_name=project_name,
status=SyncStatus.SYNCING,
message="Processing deletions",
files_processed=files_processed,
)
with logfire.span("process_deletes", delete_count=len(report.deleted)):
for path in report.deleted:
await self.handle_delete(path)
# then new and modified
for path in report.new:
entity, _ = await self.sync_file(path, new=True)
with logfire.span("process_new_files", new_count=len(report.new)):
for path in report.new:
entity, _ = await self.sync_file(path, new=True)
# Track if file was skipped
if entity is None and await self._should_skip_file(path):
failure_info = self._file_failures[path]
report.skipped_files.append(
SkippedFile(
path=path,
reason=failure_info.last_error,
failure_count=failure_info.count,
first_failed=failure_info.first_failure,
# Track if file was skipped
if entity is None and await self._should_skip_file(path):
failure_info = self._file_failures[path]
report.skipped_files.append(
SkippedFile(
path=path,
reason=failure_info.last_error,
failure_count=failure_info.count,
first_failed=failure_info.first_failure,
)
)
)
files_processed += 1
if project_name:
sync_status_tracker.update_project_progress(
project_name=project_name,
status=SyncStatus.SYNCING,
message="Processing new files",
files_processed=files_processed,
)
with logfire.span("process_modified_files", modified_count=len(report.modified)):
for path in report.modified:
entity, _ = await self.sync_file(path, new=False)
for path in report.modified:
entity, _ = await self.sync_file(path, new=False)
# Track if file was skipped
if entity is None and await self._should_skip_file(path):
failure_info = self._file_failures[path]
report.skipped_files.append(
SkippedFile(
path=path,
reason=failure_info.last_error,
failure_count=failure_info.count,
first_failed=failure_info.first_failure,
# Track if file was skipped
if entity is None and await self._should_skip_file(path):
failure_info = self._file_failures[path]
report.skipped_files.append(
SkippedFile(
path=path,
reason=failure_info.last_error,
failure_count=failure_info.count,
first_failed=failure_info.first_failure,
)
)
# Only resolve relations if there were actual changes
# If no files changed, no new unresolved relations could have been created
with logfire.span("resolve_relations"):
if report.total > 0:
await self.resolve_relations()
else:
logger.info("Skipping relation resolution - no file changes detected")
# Update scan watermark after successful sync
# Use the timestamp from sync start (not end) to ensure we catch files
# created during the sync on the next iteration
current_file_count = await self._quick_count_files(directory)
if self.entity_repository.project_id is not None:
project = await self.project_repository.find_by_id(self.entity_repository.project_id)
if project:
await self.project_repository.update(
project.id,
{
"last_scan_timestamp": sync_start_timestamp,
"last_file_count": current_file_count,
},
)
files_processed += 1
if project_name:
sync_status_tracker.update_project_progress( # pragma: no cover
project_name=project_name,
status=SyncStatus.SYNCING,
message="Processing modified files",
files_processed=files_processed,
logger.debug(
f"Updated scan watermark: timestamp={sync_start_timestamp}, "
f"file_count={current_file_count}"
)
await self.resolve_relations()
# Mark sync as completed
if project_name:
sync_status_tracker.complete_project_sync(project_name)
duration_ms = int((time.time() - start_time) * 1000)
# Record metrics for sync operation
logfire.metric_histogram("sync.duration", unit="ms").record(duration_ms)
logfire.metric_counter("sync.files.new").add(len(report.new))
logfire.metric_counter("sync.files.modified").add(len(report.modified))
logfire.metric_counter("sync.files.deleted").add(len(report.deleted))
logfire.metric_counter("sync.files.moved").add(len(report.moves))
if report.skipped_files:
logfire.metric_counter("sync.files.skipped").add(len(report.skipped_files))
# Log summary with skipped files if any
if report.skipped_files:
logger.warning(
@@ -432,75 +382,206 @@ class SyncService:
return report
@logfire.instrument()
async def scan(self, directory):
"""Scan directory for changes compared to database state."""
"""Smart scan using watermark and file count for large project optimization.
db_paths = await self.get_db_file_state()
logger.info(f"Scanning directory {directory}. Found {len(db_paths)} db paths")
Uses scan watermark tracking to dramatically reduce scan time for large projects:
- Tracks last_scan_timestamp and last_file_count in Project model
- Uses `find -newermt` for incremental scanning (only changed files)
- Falls back to full scan when deletions detected (file count decreased)
Expected performance:
- No changes: 225x faster (2s vs 450s for 1,460 files on TigrisFS)
- Few changes: 84x faster (5s vs 420s)
- Deletions: Full scan (rare, acceptable)
Architecture:
- Get current file count quickly (find | wc -l: 1.4s)
- Compare with last_file_count to detect deletions
- If no deletions: incremental scan with find -newermt (0.2s)
- Process changed files with mtime-based comparison
"""
scan_start_time = time.time()
# Track potentially moved files by checksum
scan_result = await self.scan_directory(directory)
report = SyncReport()
# First find potential new files and record checksums
# if a path is not present in the db, it could be new or could be the destination of a move
for file_path, checksum in scan_result.files.items():
if file_path not in db_paths:
report.new.add(file_path)
report.checksums[file_path] = checksum
# Get current project to check watermark
if self.entity_repository.project_id is None:
raise ValueError("Entity repository has no project_id set")
# Now detect moves and deletions
for db_path, db_checksum in db_paths.items():
local_checksum_for_db_path = scan_result.files.get(db_path)
project = await self.project_repository.find_by_id(self.entity_repository.project_id)
if project is None:
raise ValueError(f"Project not found: {self.entity_repository.project_id}")
# file not modified
if db_checksum == local_checksum_for_db_path:
pass
# Step 1: Quick file count
logger.debug("Counting files in directory")
current_count = await self._quick_count_files(directory)
logger.debug(f"Found {current_count} files in directory")
# if checksums don't match for the same path, its modified
if local_checksum_for_db_path and db_checksum != local_checksum_for_db_path:
report.modified.add(db_path)
report.checksums[db_path] = local_checksum_for_db_path
# Step 2: Determine scan strategy based on watermark and file count
if project.last_file_count is None:
# First sync ever → full scan
scan_type = "full_initial"
logger.info("First sync for this project, performing full scan")
file_paths_to_scan = await self._scan_directory_full(directory)
# check if it's moved or deleted
if not local_checksum_for_db_path:
# if we find the checksum in another file, it's a move
if db_checksum in scan_result.checksums:
new_path = scan_result.checksums[db_checksum]
report.moves[db_path] = new_path
elif current_count < project.last_file_count:
# Files deleted → need full scan to detect which ones
scan_type = "full_deletions"
logger.info(
f"File count decreased ({project.last_file_count} → {current_count}), "
f"running full scan to detect deletions"
)
file_paths_to_scan = await self._scan_directory_full(directory)
# Remove from new files if present
if new_path in report.new:
report.new.remove(new_path)
elif project.last_scan_timestamp is not None:
# Incremental scan: only files modified since last scan
scan_type = "incremental"
logger.info(
f"Running incremental scan for files modified since {project.last_scan_timestamp}"
)
file_paths_to_scan = await self._scan_directory_modified_since(
directory, project.last_scan_timestamp
)
logger.info(
f"Incremental scan found {len(file_paths_to_scan)} potentially changed files"
)
# deleted
else:
report.deleted.add(db_path)
logger.info(f"Completed scan for directory {directory}, found {report.total} changes.")
return report
else:
# Fallback to full scan (no watermark available)
scan_type = "full_fallback"
logger.warning("No scan watermark available, falling back to full scan")
file_paths_to_scan = await self._scan_directory_full(directory)
async def get_db_file_state(self) -> Dict[str, str]:
"""Get file_path and checksums from database.
Optimized to query only the columns we need (file_path, checksum) without
loading full entities or their relationships. This is 10-100x faster for
large projects compared to loading all entities with observations/relations.
Returns:
Dict mapping file paths to checksums
"""
# Query only the columns we need - no entity objects or relationships
query = select(Entity.file_path, Entity.checksum).where(
Entity.project_id == self.entity_repository.project_id
# Record scan type metric
logfire.metric_counter(f"sync.scan.{scan_type}").add(1)
logfire.metric_histogram("sync.scan.files_scanned", unit="files").record(
len(file_paths_to_scan)
)
async with db.scoped_session(self.entity_repository.session_maker) as session:
result = await session.execute(query)
rows = result.all()
# Step 3: Process each file with mtime-based comparison
scanned_paths: Set[str] = set()
changed_checksums: Dict[str, str] = {}
logger.info(f"Found {len(rows)} db file records")
return {row.file_path: row.checksum or "" for row in rows}
logger.debug(f"Processing {len(file_paths_to_scan)} files with mtime-based comparison")
for rel_path in file_paths_to_scan:
scanned_paths.add(rel_path)
# Get file stats
abs_path = directory / rel_path
if not abs_path.exists():
# File was deleted between scan and now (race condition)
continue
stat_info = abs_path.stat()
# Indexed lookup - single file query (not full table scan)
db_entity = await self.entity_repository.get_by_file_path(rel_path)
if db_entity is None:
# New file - need checksum for move detection
checksum = await self.file_service.compute_checksum(rel_path)
report.new.add(rel_path)
changed_checksums[rel_path] = checksum
logger.trace(f"New file detected: {rel_path}")
continue
# File exists in DB - check if mtime/size changed
db_mtime = db_entity.mtime
db_size = db_entity.size
fs_mtime = stat_info.st_mtime
fs_size = stat_info.st_size
# Compare mtime and size (like rsync/rclone)
# Allow small epsilon for float comparison (0.01s = 10ms)
mtime_changed = db_mtime is None or abs(fs_mtime - db_mtime) > 0.01
size_changed = db_size is None or fs_size != db_size
if mtime_changed or size_changed:
# File modified - compute checksum
checksum = await self.file_service.compute_checksum(rel_path)
db_checksum = db_entity.checksum
# Only mark as modified if checksum actually differs
# (handles cases where mtime changed but content didn't, e.g., git operations)
if checksum != db_checksum:
report.modified.add(rel_path)
changed_checksums[rel_path] = checksum
logger.trace(
f"Modified file detected: {rel_path}, "
f"mtime_changed={mtime_changed}, size_changed={size_changed}"
)
else:
# File unchanged - no checksum needed
logger.trace(f"File unchanged (mtime/size match): {rel_path}")
# Step 4: Detect moves (for both full and incremental scans)
# Check if any "new" files are actually moves by matching checksums
for new_path in list(report.new): # Use list() to allow modification during iteration
new_checksum = changed_checksums.get(new_path)
if not new_checksum:
continue
# Look for existing entity with same checksum but different path
# This could be a move or a copy
existing_entities = await self.entity_repository.find_by_checksum(new_checksum)
for candidate in existing_entities:
if candidate.file_path == new_path:
# Same path, skip (shouldn't happen for "new" files but be safe)
continue
# Check if the old path still exists on disk
old_path_abs = directory / candidate.file_path
if old_path_abs.exists():
# Original still exists → this is a copy, not a move
logger.trace(
f"File copy detected (not move): {candidate.file_path} copied to {new_path}"
)
continue
# Original doesn't exist → this is a move!
report.moves[candidate.file_path] = new_path
report.new.remove(new_path)
logger.trace(f"Move detected: {candidate.file_path} -> {new_path}")
break # Only match first candidate
# Step 5: Detect deletions (only for full scans)
# Incremental scans can't reliably detect deletions since they only see modified files
if scan_type in ("full_initial", "full_deletions", "full_fallback"):
# Use optimized query for just file paths (not full entities)
db_file_paths = await self.entity_repository.get_all_file_paths()
logger.debug(f"Found {len(db_file_paths)} db paths for deletion detection")
for db_path in db_file_paths:
if db_path not in scanned_paths:
# File in DB but not on filesystem
# Check if it was already detected as a move
if db_path in report.moves:
# Already handled as a move, skip
continue
# File was deleted
report.deleted.add(db_path)
logger.trace(f"Deleted file detected: {db_path}")
# Store checksums for files that need syncing
report.checksums = changed_checksums
scan_duration_ms = int((time.time() - scan_start_time) * 1000)
logfire.metric_histogram("sync.scan.duration", unit="ms").record(scan_duration_ms)
logger.info(
f"Completed {scan_type} scan for directory {directory} in {scan_duration_ms}ms, "
f"found {report.total} changes (new={len(report.new)}, "
f"modified={len(report.modified)}, deleted={len(report.deleted)}, "
f"moves={len(report.moves)})"
)
return report
@logfire.instrument()
async def sync_file(
self, path: str, new: bool = True
) -> Tuple[Optional[Entity], Optional[str]]:
@@ -555,6 +636,7 @@ class SyncService:
return None, None
@logfire.instrument()
async def sync_markdown_file(self, path: str, new: bool = True) -> Tuple[Optional[Entity], str]:
"""Sync a markdown file with full processing.
@@ -568,8 +650,7 @@ class SyncService:
# Parse markdown first to get any existing permalink
logger.debug(f"Parsing markdown file, path: {path}, new: {new}")
file_path = self.entity_parser.base_path / path
file_content = await self._read_file_async(file_path)
file_content = await self.file_service.read_file_content(path)
file_contains_frontmatter = has_frontmatter(file_content)
# Get file timestamps for tracking modification times
@@ -613,12 +694,20 @@ class SyncService:
# 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._compute_checksum_async(path)
final_checksum = await self.file_service.compute_checksum(path)
# Update checksum and timestamps from file system
# Update checksum, timestamps, and file metadata from file system
# Store mtime/size for efficient change detection in future scans
# This ensures temporal ordering in search and recent activity uses actual file modification times
await self.entity_repository.update(
entity.id, {"checksum": final_checksum, "created_at": created, "updated_at": modified}
entity.id,
{
"checksum": final_checksum,
"created_at": created,
"updated_at": modified,
"mtime": file_stats.st_mtime,
"size": file_stats.st_size,
},
)
logger.debug(
@@ -630,6 +719,7 @@ class SyncService:
# Return the final checksum to ensure everything is consistent
return entity, final_checksum
@logfire.instrument()
async def sync_regular_file(self, path: str, new: bool = True) -> Tuple[Optional[Entity], str]:
"""Sync a non-markdown file with basic tracking.
@@ -640,7 +730,7 @@ class SyncService:
Returns:
Tuple of (entity, checksum)
"""
checksum = await self._compute_checksum_async(path)
checksum = await self.file_service.compute_checksum(path)
if new:
# Generate permalink from path - skip conflict checks during bulk sync
await self.entity_service.resolve_permalink(path, skip_conflict_check=True)
@@ -664,6 +754,8 @@ class SyncService:
created_at=created,
updated_at=modified,
content_type=content_type,
mtime=file_stats.st_mtime,
size=file_stats.st_size,
)
)
return entity, checksum
@@ -679,8 +771,16 @@ class SyncService:
logger.error(f"Entity not found after constraint violation, path={path}")
raise ValueError(f"Entity not found after constraint violation: {path}")
# Re-get file stats since we're in update path
file_stats_for_update = self.file_service.file_stats(path)
updated = await self.entity_repository.update(
entity.id, {"file_path": path, "checksum": checksum}
entity.id,
{
"file_path": path,
"checksum": checksum,
"mtime": file_stats_for_update.st_mtime,
"size": file_stats_for_update.st_size,
},
)
if updated is None: # pragma: no cover
@@ -701,9 +801,17 @@ class SyncService:
logger.error(f"Entity not found for existing file, path={path}")
raise ValueError(f"Entity not found for existing file: {path}")
# Update checksum and modification time from file system
# Update checksum, modification time, and file metadata from file system
# Store mtime/size for efficient change detection in future scans
updated = await self.entity_repository.update(
entity.id, {"file_path": path, "checksum": checksum, "updated_at": modified}
entity.id,
{
"file_path": path,
"checksum": checksum,
"updated_at": modified,
"mtime": file_stats.st_mtime,
"size": file_stats.st_size,
},
)
if updated is None: # pragma: no cover
@@ -712,6 +820,7 @@ class SyncService:
return updated, checksum
@logfire.instrument()
async def handle_delete(self, file_path: str):
"""Handle complete entity deletion including search index cleanup."""
@@ -743,6 +852,7 @@ class SyncService:
else:
await self.search_service.delete_by_entity_id(entity.id)
@logfire.instrument()
async def handle_move(self, old_path, new_path):
logger.debug("Moving entity", old_path=old_path, new_path=new_path)
@@ -847,6 +957,7 @@ class SyncService:
# update search index
await self.search_service.index_entity(updated)
@logfire.instrument()
async def resolve_relations(self, entity_id: int | None = None):
"""Try to resolve unresolved relations.
@@ -908,64 +1019,163 @@ class SyncService:
# update search index
await self.search_service.index_entity(resolved_entity)
async def scan_directory(self, directory: Path) -> ScanResult:
async def _quick_count_files(self, directory: Path) -> int:
"""Fast file count using find command.
Uses subprocess to leverage OS-level file counting which is much faster
than Python iteration, especially on network filesystems like TigrisFS.
Args:
directory: Directory to count files in
Returns:
Number of files in directory (recursive)
"""
Scan directory for markdown files and their checksums.
process = await asyncio.create_subprocess_shell(
f'find "{directory}" -type f | wc -l',
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
error_msg = stderr.decode().strip()
logger.error(
f"FILE COUNT OPTIMIZATION FAILED: find command failed with exit code {process.returncode}, "
f"error: {error_msg}. Falling back to manual count. "
f"This will slow down watermark detection!"
)
# Track optimization failures for visibility
logfire.metric_counter("sync.scan.file_count_failure").add(1)
# Fallback: count using scan_directory
count = 0
async for _ in self.scan_directory(directory):
count += 1
return count
return int(stdout.strip())
async def _scan_directory_modified_since(
self, directory: Path, since_timestamp: float
) -> List[str]:
"""Use find -newermt for filesystem-level filtering of modified files.
This is dramatically faster than scanning all files and comparing mtimes,
especially on network filesystems like TigrisFS where stat operations are expensive.
Args:
directory: Directory to scan
since_timestamp: Unix timestamp to find files newer than
Returns:
List of relative file paths modified since the timestamp (respects .bmignore)
"""
# Convert timestamp to find-compatible format
since_date = datetime.fromtimestamp(since_timestamp).strftime("%Y-%m-%d %H:%M:%S")
process = await asyncio.create_subprocess_shell(
f'find "{directory}" -type f -newermt "{since_date}"',
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
error_msg = stderr.decode().strip()
logger.error(
f"SCAN OPTIMIZATION FAILED: find -newermt command failed with exit code {process.returncode}, "
f"error: {error_msg}. Falling back to full scan. "
f"This will cause slow syncs on large projects!"
)
# Track optimization failures for visibility
logfire.metric_counter("sync.scan.optimization_failure").add(1)
# Fallback to full scan
return await self._scan_directory_full(directory)
# Convert absolute paths to relative and filter through ignore patterns
file_paths = []
for line in stdout.decode().splitlines():
if line:
try:
abs_path = Path(line)
rel_path = abs_path.relative_to(directory).as_posix()
# Apply ignore patterns (same as scan_directory)
if should_ignore_path(abs_path, directory, self._ignore_patterns):
logger.trace(f"Ignoring path per .bmignore: {rel_path}")
continue
file_paths.append(rel_path)
except ValueError:
# Path is not relative to directory, skip it
logger.warning(f"Skipping file not under directory: {line}")
continue
return file_paths
async def _scan_directory_full(self, directory: Path) -> List[str]:
"""Full directory scan returning all file paths.
Uses scan_directory() which respects .bmignore patterns.
Args:
directory: Directory to scan
Returns:
ScanResult containing found files and any errors
List of relative file paths (respects .bmignore)
"""
start_time = time.time()
file_paths = []
async for file_path_str, _ in self.scan_directory(directory):
rel_path = Path(file_path_str).relative_to(directory).as_posix()
file_paths.append(rel_path)
return file_paths
logger.debug(f"Scanning directory {directory}")
result = ScanResult()
ignored_count = 0
async def scan_directory(self, directory: Path) -> AsyncIterator[Tuple[str, os.stat_result]]:
"""Stream files from directory using aiofiles.os.scandir() with cached stat info.
for root, dirnames, filenames in os.walk(str(directory)):
# Convert root to Path for easier manipulation
root_path = Path(root)
This method uses aiofiles.os.scandir() to leverage async I/O and cached stat
information from directory entries. This reduces network I/O by 50% on network
filesystems like TigrisFS by avoiding redundant stat() calls.
# Filter out ignored directories in-place
dirnames_to_remove = []
for dirname in dirnames:
dir_path = root_path / dirname
if should_ignore_path(dir_path, directory, self._ignore_patterns):
dirnames_to_remove.append(dirname)
ignored_count += 1
Args:
directory: Directory to scan
# Remove ignored directories from dirnames to prevent os.walk from descending
for dirname in dirnames_to_remove:
dirnames.remove(dirname)
Yields:
Tuples of (absolute_file_path, stat_info) for each file
"""
try:
entries = await aiofiles.os.scandir(directory)
except PermissionError:
logger.warning(f"Permission denied scanning directory: {directory}")
return
for filename in filenames:
path = root_path / filename
results = []
subdirs = []
# Check if file should be ignored
if should_ignore_path(path, directory, self._ignore_patterns):
ignored_count += 1
logger.trace(f"Ignoring file per .bmignore: {path.relative_to(directory)}")
continue
for entry in entries:
entry_path = Path(entry.path)
rel_path = path.relative_to(directory).as_posix()
checksum = await self._compute_checksum_async(rel_path)
result.files[rel_path] = checksum
result.checksums[checksum] = rel_path
# Check ignore patterns
if should_ignore_path(entry_path, directory, self._ignore_patterns):
logger.trace(f"Ignoring path per .bmignore: {entry_path.relative_to(directory)}")
continue
logger.trace(f"Found file, path={rel_path}, checksum={checksum}")
if entry.is_dir(follow_symlinks=False):
# Collect subdirectories to recurse into
subdirs.append(entry_path)
elif entry.is_file(follow_symlinks=False):
# Get cached stat info (no extra syscall!)
stat_info = entry.stat(follow_symlinks=False)
results.append((entry.path, stat_info))
duration_ms = int((time.time() - start_time) * 1000)
logger.debug(
f"{directory} scan completed "
f"directory={str(directory)} "
f"files_found={len(result.files)} "
f"files_ignored={ignored_count} "
f"duration_ms={duration_ms}"
)
# Yield files from current directory
for file_path, stat_info in results:
yield (file_path, stat_info)
return result
# Recurse into subdirectories
for subdir in subdirs:
async for result in self.scan_directory(subdir):
yield result
async def get_sync_service(project: Project) -> SyncService: # pragma: no cover
@@ -986,6 +1196,7 @@ async def get_sync_service(project: Project) -> SyncService: # pragma: no cover
observation_repository = ObservationRepository(session_maker, project_id=project.id)
relation_repository = RelationRepository(session_maker, project_id=project.id)
search_repository = SearchRepository(session_maker, project_id=project.id)
project_repository = ProjectRepository(session_maker)
# Initialize services
search_service = SearchService(search_repository, entity_repository, file_service)
@@ -1008,6 +1219,7 @@ async def get_sync_service(project: Project) -> SyncService: # pragma: no cover
entity_parser=entity_parser,
entity_repository=entity_repository,
relation_repository=relation_repository,
project_repository=project_repository,
search_service=search_service,
file_service=file_service,
)
src/basic_memory/sync/watch_service.py
+6 -3
View File
@@ -239,11 +239,14 @@ class WatchService:
# Check if project still exists in configuration before processing
# This prevents deleted projects from being recreated by background sync
from basic_memory.config import ConfigManager
config_manager = ConfigManager()
if project.name not in config_manager.projects and project.permalink not in config_manager.projects:
if (
project.name not in config_manager.projects
and project.permalink not in config_manager.projects
):
logger.info(
f"Skipping sync for deleted project: {project.name}, "
f"change_count={len(changes)}"
f"Skipping sync for deleted project: {project.name}, change_count={len(changes)}"
)
return
src/basic_memory/utils.py
+15
View File
@@ -185,6 +185,21 @@ def setup_logging(
logger.info(f"ENV: '{env}' Log level: '{log_level}' Logging to {log_file}")
# Bind environment context for structured logging (works in both local and cloud)
tenant_id = os.getenv("BASIC_MEMORY_TENANT_ID", "local")
fly_app_name = os.getenv("FLY_APP_NAME", "local")
fly_machine_id = os.getenv("FLY_MACHINE_ID", "local")
fly_region = os.getenv("FLY_REGION", "local")
logger.configure(
extra={
"tenant_id": tenant_id,
"fly_app_name": fly_app_name,
"fly_machine_id": fly_machine_id,
"fly_region": fly_region,
}
)
# Reduce noise from third-party libraries
noisy_loggers = {
# HTTP client logs