diff --git a/src/basic_memory/db.py b/src/basic_memory/db.py index d9f56a05..81500511 100644 --- a/src/basic_memory/db.py +++ b/src/basic_memory/db.py @@ -19,7 +19,7 @@ from sqlalchemy.ext.asyncio import ( AsyncEngine, async_scoped_session, ) -from sqlalchemy.pool import NullPool +from sqlalchemy.pool import AsyncAdaptedQueuePool, NullPool from basic_memory.repository.postgres_search_repository import PostgresSearchRepository from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository @@ -216,19 +216,31 @@ def _create_sqlite_engine(db_url: str, db_type: DatabaseType) -> AsyncEngine: "isolation_level": None, # Use autocommit mode } ) + + if db_type == DatabaseType.MEMORY: + # Trigger: an in-memory SQLite URL would default to StaticPool, which hands the + # same DBAPI connection to every concurrently checked-out session. + # Why: concurrent asyncio tasks then share one transaction scope — a rollback + # issued by one session (scoped_session exception handling or the pool's + # reset-on-return) silently destroys another session's uncommitted writes (#940). + # Outcome: a single-connection blocking queue pool keeps the in-memory database + # alive for the engine's lifetime while serializing sessions at transaction + # granularity, restoring the isolation the repositories assume. + engine = create_async_engine( + db_url, + connect_args=connect_args, + poolclass=AsyncAdaptedQueuePool, + pool_size=1, + max_overflow=0, + ) + elif os.name == "nt": # Use NullPool for Windows filesystem databases to avoid connection pooling issues - # Important: Do NOT use NullPool for in-memory databases as it will destroy the database - # between connections - if db_type == DatabaseType.FILESYSTEM: - engine = create_async_engine( - db_url, - connect_args=connect_args, - poolclass=NullPool, # Disable connection pooling on Windows - echo=False, - ) - else: - # In-memory databases need connection pooling to maintain state - engine = create_async_engine(db_url, connect_args=connect_args) + engine = create_async_engine( + db_url, + connect_args=connect_args, + poolclass=NullPool, # Disable connection pooling on Windows + echo=False, + ) else: engine = create_async_engine(db_url, connect_args=connect_args) diff --git a/tests/db/test_memory_db_session_isolation.py b/tests/db/test_memory_db_session_isolation.py new file mode 100644 index 00000000..1e25b4d9 --- /dev/null +++ b/tests/db/test_memory_db_session_isolation.py @@ -0,0 +1,93 @@ +"""Regression test for issue #940: lost writes on the in-memory SQLite engine. + +The in-memory SQLite URL (``sqlite+aiosqlite://``) used to fall back to +SQLAlchemy's StaticPool, which hands the same DBAPI connection to every +concurrently checked-out session. Concurrent asyncio tasks then share one +transaction scope: a rollback issued by one session — scoped_session's +exception handling, or the pool's reset-on-return at checkin — silently rolls +back another session's uncommitted writes. During sync this manifested as a +freshly inserted relation row vanishing without any error, which is how +``test_sync_entity_circular_relations`` failed on CI with +``len(entity_b.outgoing_relations) == 0``. + +This test pins the isolation invariant directly: a session that rolls back in +one task must never destroy an uncommitted write of a session in another task. +""" + +import asyncio +from pathlib import Path + +import pytest +from sqlalchemy import text + +from basic_memory import db +from basic_memory.models import Base + + +class _SimulatedIndexingFailure(Exception): + """Stand-in for any per-file error that _run_bounded swallows during sync.""" + + +@pytest.mark.asyncio +async def test_concurrent_session_rollback_does_not_destroy_uncommitted_writes(): + """A rolled-back session in one task must not erase another task's writes.""" + async with db.engine_session_factory( + db_path=Path("unused.db"), db_type=db.DatabaseType.MEMORY + ) as (engine, session_maker): + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + # Seed a project and entity so a relation row satisfies its FK constraints. + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "INSERT INTO project (id, external_id, name, description, path, is_active," + " is_default, created_at, updated_at, permalink) " + "VALUES (1, 'px', 'p', '', '/tmp', 1, 1, '2024-01-01', '2024-01-01', 'p')" + ) + ) + await session.execute( + text( + "INSERT INTO entity (id, external_id, title, note_type, content_type," + " project_id, file_path, created_at, updated_at) " + "VALUES (1, 'ex', 'E', 'note', 'text/markdown', 1, 'e.md'," + " '2024-01-01', '2024-01-01')" + ) + ) + + write_in_flight = asyncio.Event() + + async def writer() -> None: + # Mirrors RelationRepository.add_all_ignore_duplicates: INSERT executed, + # commit only happens at scoped_session exit several awaits later. + async with db.scoped_session(session_maker) as session: + await session.execute( + text( + "INSERT INTO relation (project_id, from_id, to_id, to_name," + " relation_type) VALUES (1, 1, NULL, 'target', 'depends_on')" + ) + ) + write_in_flight.set() + # Real DB roundtrips (not sleeps) keep this transaction open across + # await points, exactly like the multi-statement sessions in sync. + for _ in range(10): + await session.execute(text("SELECT 1")) + + async def failing_reader() -> None: + # Mirrors any per-file failure during batch indexing: scoped_session + # rolls back on the exception path while sibling tasks are mid-write. + await write_in_flight.wait() + with pytest.raises(_SimulatedIndexingFailure): + async with db.scoped_session(session_maker) as session: + await session.execute(text("SELECT 1")) + raise _SimulatedIndexingFailure() + + await asyncio.gather(writer(), failing_reader()) + + async with db.scoped_session(session_maker) as session: + count = (await session.execute(text("SELECT count(*) FROM relation"))).scalar() + + assert count == 1, ( + "writer's committed INSERT was rolled back by a concurrent session — " + "the in-memory engine is sharing one transaction scope across sessions" + ) diff --git a/tests/repository/test_entity_repository.py b/tests/repository/test_entity_repository.py index 194722af..8464b42c 100644 --- a/tests/repository/test_entity_repository.py +++ b/tests/repository/test_entity_repository.py @@ -291,9 +291,11 @@ async def test_delete_entity_with_relations(entity_repository: EntityRepository, remaining_relations = result.scalars().all() assert len(remaining_relations) == 0 - # Verify target entity still exists - target_exists = await entity_repository.find_by_id(target.id) - assert target_exists is not None + # Verify target entity still exists. Runs outside the session block above: + # find_by_id opens its own scoped session, and the serialized in-memory pool + # (one connection, see #940) deadlocks on nested session checkouts. + target_exists = await entity_repository.find_by_id(target.id) + assert target_exists is not None @pytest.mark.asyncio