From c50d97e548759e5f513cb03f95696ec9d64ffb24 Mon Sep 17 00:00:00 2001 From: phernandez Date: Thu, 16 Apr 2026 18:04:17 -0500 Subject: [PATCH] fix(sync): instrument single markdown indexing Signed-off-by: phernandez --- src/basic_memory/indexing/batch_indexer.py | 38 +++++++++++----- tests/sync/test_sync_service_telemetry.py | 53 ++++++++++++++++++++++ 2 files changed, 79 insertions(+), 12 deletions(-) diff --git a/src/basic_memory/indexing/batch_indexer.py b/src/basic_memory/indexing/batch_indexer.py index 2535d6c0..0ec655b1 100644 --- a/src/basic_memory/indexing/batch_indexer.py +++ b/src/basic_memory/indexing/batch_indexer.py @@ -11,6 +11,7 @@ from typing import Awaitable, Callable, Mapping, TypeVar from loguru import logger from sqlalchemy.exc import IntegrityError +from basic_memory import telemetry from basic_memory.config import BasicMemoryConfig from basic_memory.file_utils import compute_checksum, has_frontmatter, remove_frontmatter from basic_memory.markdown.schemas import EntityMarkdown @@ -190,35 +191,48 @@ class BatchIndexer: if not self._is_markdown(file): raise ValueError(f"index_markdown_file requires markdown input: {file.path}") - prepared = await self._prepare_markdown_file(file) + with telemetry.span("index.markdown_file.prepare", path=file.path): + prepared = await self._prepare_markdown_file(file) if existing_permalink_by_path is None: - existing_permalink_by_path = { - path: permalink - for path, permalink in ( - await self.entity_repository.get_file_path_to_permalink_map() - ).items() - } + with telemetry.span("index.markdown_file.load_permalink_map", path=file.path): + existing_permalink_by_path = { + path: permalink + for path, permalink in ( + await self.entity_repository.get_file_path_to_permalink_map() + ).items() + } reserved_permalinks = { permalink for path, permalink in existing_permalink_by_path.items() if path != file.path and permalink } - prepared = await self._normalize_markdown_file(prepared, reserved_permalinks) + with telemetry.span("index.markdown_file.normalize", path=file.path): + prepared = await self._normalize_markdown_file(prepared, reserved_permalinks) existing_permalink_by_path[file.path] = prepared.markdown.frontmatter.permalink - persisted = await self._persist_markdown_file(prepared, is_new=new) + with telemetry.span("index.markdown_file.persist", path=file.path, is_new=new): + persisted = await self._persist_markdown_file(prepared, is_new=new) existing_permalink_by_path[file.path] = persisted.entity.permalink - await self._resolve_batch_relations([persisted.entity.id], max_concurrent=1) - refreshed = await self.entity_repository.find_by_ids([persisted.entity.id]) + with telemetry.span( + "index.markdown_file.reload_entity", + path=file.path, + entity_id=persisted.entity.id, + ): + refreshed = await self.entity_repository.find_by_ids([persisted.entity.id]) if len(refreshed) != 1: # pragma: no cover raise ValueError(f"Failed to reload indexed entity for {file.path}") entity = refreshed[0] prepared_entity = self._build_prepared_entity(persisted.prepared, entity) if index_search: - return await self._refresh_search_index(prepared_entity, entity) + with telemetry.span( + "index.markdown_file.refresh_search_index", + path=file.path, + entity_id=entity.id, + ): + return await self._refresh_search_index(prepared_entity, entity) return IndexedEntity( path=prepared_entity.path, diff --git a/tests/sync/test_sync_service_telemetry.py b/tests/sync/test_sync_service_telemetry.py index 9a657508..1f59964b 100644 --- a/tests/sync/test_sync_service_telemetry.py +++ b/tests/sync/test_sync_service_telemetry.py @@ -4,12 +4,16 @@ from __future__ import annotations import importlib from contextlib import contextmanager +from pathlib import Path +from textwrap import dedent from types import SimpleNamespace +from unittest.mock import AsyncMock import pytest from basic_memory.sync.sync_service import SyncReport +batch_indexer_module = importlib.import_module("basic_memory.indexing.batch_indexer") sync_service_module = importlib.import_module("basic_memory.sync.sync_service") @@ -30,6 +34,14 @@ def _capture_sync_telemetry(): return operations, spans, fake_operation, fake_span +def _write_markdown(project_root: Path, relative_path: str, content: str) -> Path: + """Create one markdown file under the test project.""" + file_path = project_root / relative_path + file_path.parent.mkdir(parents=True, exist_ok=True) + file_path.write_text(content, encoding="utf-8") + return file_path + + @pytest.mark.asyncio async def test_sync_emits_phase_spans(sync_service, project_config, monkeypatch) -> None: operations, spans, fake_operation, fake_span = _capture_sync_telemetry() @@ -97,6 +109,47 @@ async def test_sync_emits_phase_spans(sync_service, project_config, monkeypatch) ] +@pytest.mark.asyncio +async def test_sync_one_markdown_file_emits_index_phase_spans( + sync_service, test_project, monkeypatch +) -> None: + _, spans, _, fake_span = _capture_sync_telemetry() + monkeypatch.setattr(batch_indexer_module.telemetry, "span", fake_span) + monkeypatch.setattr(sync_service.search_service, "index_entity_data", AsyncMock()) + + _write_markdown( + Path(test_project.path), + "notes/telemetry.md", + dedent( + f"""\ + --- + title: Telemetry Note + type: note + permalink: {test_project.name}/notes/telemetry + --- + + # Telemetry Note + + Body content. + """ + ), + ) + + result = await sync_service.sync_one_markdown_file("notes/telemetry.md") + + index_spans = [(name, attrs) for name, attrs in spans if name.startswith("index.markdown_file")] + assert [name for name, _ in index_spans] == [ + "index.markdown_file.prepare", + "index.markdown_file.load_permalink_map", + "index.markdown_file.normalize", + "index.markdown_file.persist", + "index.markdown_file.reload_entity", + ] + assert index_spans[0][1] == {"path": "notes/telemetry.md"} + assert index_spans[3][1] == {"path": "notes/telemetry.md", "is_new": True} + assert index_spans[4][1]["entity_id"] == result.entity.id + + @pytest.mark.asyncio async def test_sync_file_emits_failure_span(sync_service, monkeypatch) -> None: _, spans, _, fake_span = _capture_sync_telemetry()