test-int/test_sqlite_lifecycle_integration.py
Rama main · réplica de lectura del repositorio original.
"""Real file-backed SQLite contention and CLI-exit probes for #1430/#1322.
No platform simulation: Windows CI selects the production Windows engine path.
The CLI probe uses real FastEmbed inference to cover semantic reindex shutdown.
"""
import asyncio
import json
import os
from pathlib import Path
import subprocess
import sys
import time
import pytest
from sqlalchemy import event, text
from basic_memory.config import BasicMemoryConfig, DatabaseBackend, ProjectEntry
from basic_memory.db import DatabaseType, _create_sqlite_engine
@pytest.mark.asyncio
async def test_file_sqlite_concurrent_writes_commit_and_dispose(tmp_path: Path) -> None:
# An explicit config keeps the developer's home config from choosing the backend.
app_config = BasicMemoryConfig(
env="test",
projects={"test-project": ProjectEntry(path=str(tmp_path))},
default_project="test-project",
database_backend=DatabaseBackend.SQLITE,
)
engine = _create_sqlite_engine(
DatabaseType.get_db_url(tmp_path / "contention.db", DatabaseType.FILESYSTEM, app_config),
DatabaseType.FILESYSTEM,
app_config,
)
connections = 0
@event.listens_for(engine.sync_engine, "connect")
def count_connection(connection: object, record: object) -> None:
nonlocal connections
connections += 1
started = time.monotonic()
try:
async with engine.begin() as connection:
assert (await connection.execute(text("PRAGMA journal_mode"))).scalar() == "wal"
assert (await connection.execute(text("PRAGMA busy_timeout"))).scalar() == 10000
await connection.execute(text("CREATE TABLE notes (id INTEGER PRIMARY KEY, body TEXT)"))
await connection.execute(text("CREATE VIRTUAL TABLE notes_fts USING fts5(body)"))
async def write_note(note_id: int) -> None:
async with engine.begin() as connection:
await connection.execute(
text("INSERT INTO notes VALUES (:id, :body)"),
{"id": note_id, "body": "searchable note"},
)
await connection.execute(
text("INSERT INTO notes_fts(rowid, body) VALUES (:id, :body)"),
{"id": note_id, "body": "searchable note"},
)
assert (
await connection.execute(
text("SELECT body FROM notes WHERE id = :id"), {"id": note_id}
)
).scalar_one() == "searchable note"
# Trigger: 64 writers share the pool's connections, and every round they all
# want SQLite's single write lock at once.
# Why: SQLite's busy handler polls with growing sleeps (up to 100 ms) instead of
# queueing waiters. When writers ran 40 transactions back to back, the other
# pooled connections could hand the lock between themselves for the whole run
# while one waiter kept sleeping through every free moment; on a slow Windows
# runner that waiter passed busy_timeout (10 s) and raised "database is
# locked" (#1955). Nothing in production holds the write lock saturated for
# 20 s, and the busy handler never promised fairness.
# Outcome: each round ends when its 64 writes commit, so a waiting connection
# competes only with that round's work (well under a second), never with the
# rest of the run. Contention per round is unchanged.
async with asyncio.timeout(90):
for index in range(40):
async with asyncio.TaskGroup() as writers:
for writer in range(64):
writers.create_task(write_note(writer * 40 + index))
async with engine.connect() as connection:
assert (await connection.execute(text("SELECT count(*) FROM notes"))).scalar() == 2560
assert (
await connection.execute(
text("SELECT count(*) FROM notes_fts WHERE notes_fts MATCH 'searchable'")
)
).scalar() == 2560
assert (await connection.execute(text("PRAGMA integrity_check"))).scalar() == "ok"
finally:
await asyncio.wait_for(engine.dispose(), 15)
print(
f"pool={type(engine.pool).__name__} connections={connections} "
f"elapsed={time.monotonic() - started:.2f}s"
)
@pytest.mark.semantic
def test_semantic_reindex_subprocess_exits_after_full_and_incremental_runs(tmp_path: Path) -> None:
project = tmp_path / "notes"
project.mkdir()
for index in range(15):
(project / f"note-{index}.md").write_text(
f"# Note {index}\n\n- [fact] searchable lifecycle probe\n", encoding="utf-8"
)
config_dir = tmp_path / "config"
config_dir.mkdir()
(config_dir / "config.json").write_text(
json.dumps(
{
"projects": {"probe": {"path": str(project), "mode": "local"}},
"default_project": "probe",
"semantic_embedding_provider": "fastembed",
"semantic_embedding_model": "BAAI/bge-small-en-v1.5",
"auto_update": False,
}
),
encoding="utf-8",
)
environment = {
key: value
for key, value in os.environ.items()
if not key.startswith("BASIC_MEMORY_") and key != "PYTEST_CURRENT_TEST"
}
environment["BASIC_MEMORY_CONFIG_DIR"] = str(config_dir)
environment["PYTHONUTF8"] = "1"
# Native thread stacks survive in the CI artifact if the child wedges.
entrypoint = (
"import faulthandler; faulthandler.dump_traceback_later(60); "
"from basic_memory.cli.main import app; app()"
)
for arguments in (["reindex", "--full"], ["reindex"]):
try:
result = subprocess.run(
[sys.executable, "-c", entrypoint, *arguments],
env=environment,
capture_output=True,
text=True,
encoding="utf-8",
timeout=90,
)
except subprocess.TimeoutExpired as error:
pytest.fail(f"CLI did not exit: {arguments}\n{error.stdout}\n{error.stderr}")
print(result.stdout)
print(result.stderr)
assert result.returncode == 0, result.stdout + result.stderr
assert "Embeddings complete" in result.stdout
assert "BAAI/bge-small-en-v1.5" in result.stdout
assert "index=sqlite-vec" in result.stdout
assert "0 errors" in result.stdout
# The CLI's embedded total includes unchanged entities; skipped tells
# us whether the incremental run actually had new inference to do.
summary = " ".join(result.stdout.split())
assert "15 entities embedded" in summary
expected_skipped = 0 if "--full" in arguments else 14
assert f" {expected_skipped} skipped" in summary
assert "Reindex complete!" in result.stdout
# Make incremental reindex perform inference too, rather than passing
# by skipping every unchanged note without loading the ONNX model.
if "--full" in arguments:
(project / "note-0.md").write_text(
"# Note 0\n\n- [fact] updated semantic lifecycle probe\n", encoding="utf-8"
)