Skip to content

Commit 85c701b

Browse files
phernandezclaude
andcommitted
fix(sync): serialize in-memory SQLite sessions so concurrent rollbacks cannot destroy writes
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>
1 parent 1ad3a35 commit 85c701b

3 files changed

Lines changed: 123 additions & 16 deletions

File tree

src/basic_memory/db.py

Lines changed: 25 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,7 @@
1919
AsyncEngine,
2020
async_scoped_session,
2121
)
22-
from sqlalchemy.pool import NullPool
22+
from sqlalchemy.pool import AsyncAdaptedQueuePool, NullPool
2323

2424
from basic_memory.repository.postgres_search_repository import PostgresSearchRepository
2525
from basic_memory.repository.sqlite_search_repository import SQLiteSearchRepository
@@ -216,19 +216,31 @@ def _create_sqlite_engine(db_url: str, db_type: DatabaseType) -> AsyncEngine:
216216
"isolation_level": None, # Use autocommit mode
217217
}
218218
)
219+
220+
if db_type == DatabaseType.MEMORY:
221+
# Trigger: an in-memory SQLite URL would default to StaticPool, which hands the
222+
# same DBAPI connection to every concurrently checked-out session.
223+
# Why: concurrent asyncio tasks then share one transaction scope — a rollback
224+
# issued by one session (scoped_session exception handling or the pool's
225+
# reset-on-return) silently destroys another session's uncommitted writes (#940).
226+
# Outcome: a single-connection blocking queue pool keeps the in-memory database
227+
# alive for the engine's lifetime while serializing sessions at transaction
228+
# granularity, restoring the isolation the repositories assume.
229+
engine = create_async_engine(
230+
db_url,
231+
connect_args=connect_args,
232+
poolclass=AsyncAdaptedQueuePool,
233+
pool_size=1,
234+
max_overflow=0,
235+
)
236+
elif os.name == "nt":
219237
# Use NullPool for Windows filesystem databases to avoid connection pooling issues
220-
# Important: Do NOT use NullPool for in-memory databases as it will destroy the database
221-
# between connections
222-
if db_type == DatabaseType.FILESYSTEM:
223-
engine = create_async_engine(
224-
db_url,
225-
connect_args=connect_args,
226-
poolclass=NullPool, # Disable connection pooling on Windows
227-
echo=False,
228-
)
229-
else:
230-
# In-memory databases need connection pooling to maintain state
231-
engine = create_async_engine(db_url, connect_args=connect_args)
238+
engine = create_async_engine(
239+
db_url,
240+
connect_args=connect_args,
241+
poolclass=NullPool, # Disable connection pooling on Windows
242+
echo=False,
243+
)
232244
else:
233245
engine = create_async_engine(db_url, connect_args=connect_args)
234246

Lines changed: 93 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,93 @@
1+
"""Regression test for issue #940: lost writes on the in-memory SQLite engine.
2+
3+
The in-memory SQLite URL (``sqlite+aiosqlite://``) used to fall back to
4+
SQLAlchemy's StaticPool, which hands the same DBAPI connection to every
5+
concurrently checked-out session. Concurrent asyncio tasks then share one
6+
transaction scope: a rollback issued by one session — scoped_session's
7+
exception handling, or the pool's reset-on-return at checkin — silently rolls
8+
back another session's uncommitted writes. During sync this manifested as a
9+
freshly inserted relation row vanishing without any error, which is how
10+
``test_sync_entity_circular_relations`` failed on CI with
11+
``len(entity_b.outgoing_relations) == 0``.
12+
13+
This test pins the isolation invariant directly: a session that rolls back in
14+
one task must never destroy an uncommitted write of a session in another task.
15+
"""
16+
17+
import asyncio
18+
from pathlib import Path
19+
20+
import pytest
21+
from sqlalchemy import text
22+
23+
from basic_memory import db
24+
from basic_memory.models import Base
25+
26+
27+
class _SimulatedIndexingFailure(Exception):
28+
"""Stand-in for any per-file error that _run_bounded swallows during sync."""
29+
30+
31+
@pytest.mark.asyncio
32+
async def test_concurrent_session_rollback_does_not_destroy_uncommitted_writes():
33+
"""A rolled-back session in one task must not erase another task's writes."""
34+
async with db.engine_session_factory(
35+
db_path=Path("unused.db"), db_type=db.DatabaseType.MEMORY
36+
) as (engine, session_maker):
37+
async with engine.begin() as conn:
38+
await conn.run_sync(Base.metadata.create_all)
39+
40+
# Seed a project and entity so a relation row satisfies its FK constraints.
41+
async with db.scoped_session(session_maker) as session:
42+
await session.execute(
43+
text(
44+
"INSERT INTO project (id, external_id, name, description, path, is_active,"
45+
" is_default, created_at, updated_at, permalink) "
46+
"VALUES (1, 'px', 'p', '', '/tmp', 1, 1, '2024-01-01', '2024-01-01', 'p')"
47+
)
48+
)
49+
await session.execute(
50+
text(
51+
"INSERT INTO entity (id, external_id, title, note_type, content_type,"
52+
" project_id, file_path, created_at, updated_at) "
53+
"VALUES (1, 'ex', 'E', 'note', 'text/markdown', 1, 'e.md',"
54+
" '2024-01-01', '2024-01-01')"
55+
)
56+
)
57+
58+
write_in_flight = asyncio.Event()
59+
60+
async def writer() -> None:
61+
# Mirrors RelationRepository.add_all_ignore_duplicates: INSERT executed,
62+
# commit only happens at scoped_session exit several awaits later.
63+
async with db.scoped_session(session_maker) as session:
64+
await session.execute(
65+
text(
66+
"INSERT INTO relation (project_id, from_id, to_id, to_name,"
67+
" relation_type) VALUES (1, 1, NULL, 'target', 'depends_on')"
68+
)
69+
)
70+
write_in_flight.set()
71+
# Real DB roundtrips (not sleeps) keep this transaction open across
72+
# await points, exactly like the multi-statement sessions in sync.
73+
for _ in range(10):
74+
await session.execute(text("SELECT 1"))
75+
76+
async def failing_reader() -> None:
77+
# Mirrors any per-file failure during batch indexing: scoped_session
78+
# rolls back on the exception path while sibling tasks are mid-write.
79+
await write_in_flight.wait()
80+
with pytest.raises(_SimulatedIndexingFailure):
81+
async with db.scoped_session(session_maker) as session:
82+
await session.execute(text("SELECT 1"))
83+
raise _SimulatedIndexingFailure()
84+
85+
await asyncio.gather(writer(), failing_reader())
86+
87+
async with db.scoped_session(session_maker) as session:
88+
count = (await session.execute(text("SELECT count(*) FROM relation"))).scalar()
89+
90+
assert count == 1, (
91+
"writer's committed INSERT was rolled back by a concurrent session — "
92+
"the in-memory engine is sharing one transaction scope across sessions"
93+
)

tests/repository/test_entity_repository.py

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -291,9 +291,11 @@ async def test_delete_entity_with_relations(entity_repository: EntityRepository,
291291
remaining_relations = result.scalars().all()
292292
assert len(remaining_relations) == 0
293293

294-
# Verify target entity still exists
295-
target_exists = await entity_repository.find_by_id(target.id)
296-
assert target_exists is not None
294+
# Verify target entity still exists. Runs outside the session block above:
295+
# find_by_id opens its own scoped session, and the serialized in-memory pool
296+
# (one connection, see #940) deadlocks on nested session checkouts.
297+
target_exists = await entity_repository.find_by_id(target.id)
298+
assert target_exists is not None
297299

298300

299301
@pytest.mark.asyncio

0 commit comments

Comments
 (0)