mirror of
https://github.com/basicmachines-co/basic-memory
synced 2026-06-21 13:47:35 +00:00
phernandez
85c701b8c2
Root cause of the test_sync_entity_circular_relations CI failure (#940, len(entity_b.outgoing_relations) == 0): the in-memory SQLite URL (sqlite+aiosqlite://) falls back to SQLAlchemy's StaticPool, which hands the same DBAPI connection to every concurrently checked-out session. Concurrent asyncio tasks therefore share one SQLite transaction scope. A rollback issued through one session — scoped_session's exception handler, or the pool's reset-on-return at connection checkin (~40 real ROLLBACKs per sync, measured) — also rolls back any other task's executed-but-uncommitted statements. During batch indexing, per-file tasks run concurrently (asyncio.gather bounded by index_entity_max_concurrent). If a sibling session's checkin ROLLBACK lands between one task's relation INSERT and its COMMIT, the relation row is silently erased: no error is raised, the sync reports success. The final sync-level resolve_relations() pass cannot heal this because it only re-queries rows with to_id IS NULL — the destroyed INSERT leaves no row at all. Fix: give MEMORY-type engines a single-connection AsyncAdaptedQueuePool (pool_size=1, max_overflow=0). The lone connection keeps the in-memory database alive for the engine's lifetime (the reason StaticPool was used), while the blocking checkout serializes sessions at transaction granularity, restoring the isolation the repositories assume. File-based SQLite and Postgres engines are unchanged; production never uses MEMORY engines. The regression test pins the invariant directly: a session that rolls back in one task must never destroy another task's uncommitted writes. It fails deterministically against StaticPool and passes with the serialized pool. tests/repository/test_entity_repository.py held a scoped session open while calling a repository method that opens its own session — tolerated on a shared connection, a deadlock under a serialized pool — so the nested call moved out of the session block. Refs #940 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Signed-off-by: phernandez <paul@basicmachines.co>
94 lines
4.2 KiB
Python
94 lines
4.2 KiB
Python
"""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"
|
|
)
|