Skip to content

Fix SQLite concurrency risk in create_or_get method #8

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
krishna3554 wants to merge 3 commits into from
Closed

Fix SQLite concurrency risk in create_or_get method #8

Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
6952274
Add retry logic for SQLite concurrency issues in create_or_get method
krishna3554 Aug 19, 2026
5823299
Remove orphaned duplicate transaction body in create_or_get
devin-ai-integration[bot] Aug 28, 2026
bfe8a2a
Merge branch 'main' into agent/add-memory-evaluation-benchmark
krishna3554 Aug 28, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
148 changes: 80 additions & 68 deletions services/dma-api/src/dma_api/repository.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,77 +141,89 @@ def initialize(self) -> None:

def create_or_get(self, record: MemoryRecord, idempotency_key: str) -> tuple[MemoryRecord, bool]:
"""Create a record, or return the result of an exact idempotent replay."""
with self._transaction() as connection:
connection.execute("BEGIN IMMEDIATE")
existing = connection.execute(
"""
SELECT memory_id FROM idempotency_keys
WHERE tenant_id = ? AND operation = 'remember' AND idempotency_key = ?
""",
(record.tenant_id, idempotency_key),
).fetchone()
if existing is not None:
return self._get_by_id(connection, existing["memory_id"]), False

if record.type is MemoryType.SEMANTIC:
duplicate = self._find_normalized_semantic(connection, record)
if duplicate is not None:
updated = MemoryRecord(
id=duplicate.id,
tenant_id=duplicate.tenant_id,
agent_id=duplicate.agent_id,
content=record.content,
type=record.type,
version=duplicate.version + 1,
created_at=duplicate.created_at,
updated_at=record.updated_at,
expires_at=record.expires_at,
metadata=record.metadata,
max_retries = 3
for attempt in range(max_retries):
try:
with self._connect() as connection:
connection.execute("BEGIN IMMEDIATE")
Comment on lines +147 to +148

@devin-ai-integration devin-ai-integration Bot Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🟡 Failed write attempts leave database connections open, and repeated retries multiply the leak

Each attempt opens a new database connection that is never closed (with self._connect() as connection at services/dma-api/src/dma_api/repository.py:142), so retrying a locked write accumulates several open handles per request instead of one.
Impact: Under sustained concurrent writes the service can exhaust file handles and lock the database for longer, making writes fail more often.

sqlite3 connection context manager commits/rolls back but does not close

SQLiteMemoryRepository._connect (services/dma-api/src/dma_api/repository.py:403-407) returns a raw sqlite3.Connection. Using it as a context manager only manages the transaction — on exit it commits or rolls back but leaves the connection (and its file lock, until garbage collection) open. Previously one connection leaked per call; with the new loop up to max_retries connections leak per call, and the leaked connection from a failed attempt may still hold a reserved lock while the next attempt runs BEGIN IMMEDIATE, which can make the retry itself hit "database is locked". Wrapping with contextlib.closing(...) (or a try/finally close) fixes it.

Prompt for agents 
services/dma-api/src/dma_api/repository.py:_connect returns a bare sqlite3.Connection and all call sites use `with self._connect() as connection`, which manages the transaction but never closes the connection. The new retry loop in create_or_get makes this worse because a failed attempt's connection stays open (potentially still holding locks) while the next attempt begins. Consider making _connect a contextmanager that closes the connection on exit (or wrap with contextlib.closing at each call site), and confirm the retry path releases resources between attempts.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

existing = connection.execute(
"""
SELECT memory_id FROM idempotency_keys
WHERE tenant_id = ? AND operation = 'remember' AND idempotency_key = ?
""",
(record.tenant_id, idempotency_key),
).fetchone()
if existing is not None:
return self._get_by_id(connection, existing["memory_id"]), False

if record.type is MemoryType.SEMANTIC:
duplicate = self._find_normalized_semantic(connection, record)
if duplicate is not None:
updated = MemoryRecord(
id=duplicate.id,
tenant_id=duplicate.tenant_id,
agent_id=duplicate.agent_id,
content=record.content,
type=record.type,
version=duplicate.version + 1,
created_at=duplicate.created_at,
updated_at=record.updated_at,
expires_at=record.expires_at,
metadata=record.metadata,
)
connection.execute(
"""UPDATE memories SET content = ?, version = ?, updated_at = ?, expires_at = ?, metadata_json = ? WHERE id = ?""",
(updated.content, updated.version, updated.updated_at.isoformat(), updated.expires_at.isoformat() if updated.expires_at else None, json.dumps(updated.metadata, separators=(",", ":"), sort_keys=True), updated.id),
)
connection.execute("DELETE FROM memory_search WHERE memory_id = ?", (updated.id,))
connection.execute("INSERT INTO memory_search (content, memory_id, tenant_id, agent_id, type) VALUES (?, ?, ?, ?, ?)", (updated.content, updated.id, updated.tenant_id, updated.agent_id, updated.type.value))
connection.execute("INSERT INTO idempotency_keys (tenant_id, operation, idempotency_key, memory_id) VALUES (?, 'remember', ?, ?)", (record.tenant_id, idempotency_key, updated.id))
return updated, False

connection.execute(
"""
INSERT INTO memories (
id, tenant_id, agent_id, content, type, version,
created_at, updated_at, expires_at, metadata_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
record.id,
record.tenant_id,
record.agent_id,
record.content,
record.type.value,
record.version,
record.created_at.isoformat(),
record.updated_at.isoformat(),
record.expires_at.isoformat() if record.expires_at else None,
json.dumps(record.metadata, separators=(",", ":"), sort_keys=True),
),
)
connection.execute(
"""UPDATE memories SET content = ?, version = ?, updated_at = ?, expires_at = ?, metadata_json = ? WHERE id = ?""",
(updated.content, updated.version, self._utc_isoformat(updated.updated_at), self._utc_isoformat(updated.expires_at) if updated.expires_at else None, json.dumps(updated.metadata, separators=(",", ":"), sort_keys=True), updated.id),
"""
INSERT INTO memory_search (content, memory_id, tenant_id, agent_id, type)
VALUES (?, ?, ?, ?, ?)
""",
(record.content, record.id, record.tenant_id, record.agent_id, record.type.value),
)
connection.execute("DELETE FROM memory_search WHERE memory_id = ?", (updated.id,))
connection.execute("INSERT INTO memory_search (content, memory_id, tenant_id, agent_id, type) VALUES (?, ?, ?, ?, ?)", (updated.content, updated.id, updated.tenant_id, updated.agent_id, updated.type.value))
connection.execute("INSERT INTO idempotency_keys (tenant_id, operation, idempotency_key, memory_id) VALUES (?, 'remember', ?, ?)", (record.tenant_id, idempotency_key, updated.id))
return updated, False

connection.execute(
"""
INSERT INTO memories (
id, tenant_id, agent_id, content, type, version,
created_at, updated_at, expires_at, metadata_json
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
record.id,
record.tenant_id,
record.agent_id,
record.content,
record.type.value,
record.version,
self._utc_isoformat(record.created_at),
self._utc_isoformat(record.updated_at),
self._utc_isoformat(record.expires_at) if record.expires_at else None,
json.dumps(record.metadata, separators=(",", ":"), sort_keys=True),
),
)
connection.execute(
"""
INSERT INTO memory_search (content, memory_id, tenant_id, agent_id, type)
VALUES (?, ?, ?, ?, ?)
""",
(record.content, record.id, record.tenant_id, record.agent_id, record.type.value),
)
connection.execute(
"""
INSERT INTO idempotency_keys (tenant_id, operation, idempotency_key, memory_id)
VALUES (?, 'remember', ?, ?)
""",
(record.tenant_id, idempotency_key, record.id),
)
return record, True
connection.execute(
"""
INSERT INTO idempotency_keys (tenant_id, operation, idempotency_key, memory_id)
VALUES (?, 'remember', ?, ?)
""",
(record.tenant_id, idempotency_key, record.id),
)
return record, True
except sqlite3.OperationalError as error:
if "database is locked" in str(error):
if attempt == max_retries - 1:
raise
import time
time.sleep(0.1 * (2 ** attempt))
continue
raise
raise RuntimeError("Unreachable code reached")

@devin-ai-integration devin-ai-integration Bot Aug 19, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 Leftover duplicated code block after the rewritten save routine breaks the memory service entirely

An old copy of the save logic was left behind after the new retry loop's final error (raise RuntimeError("Unreachable code reached") at services/dma-api/src/dma_api/repository.py:221) at a deeper indentation than the surrounding code, so the memory service can no longer start or store anything.
Impact: The API fails to load, so every request — remembering, recalling, listing, deleting — stops working.

Incomplete refactor leaves an orphaned, unreachable copy of the transaction body

The PR moved the body of create_or_get into a for attempt in range(max_retries) / try / with self._connect() block (lines 139-221), but the pre-existing body was not deleted. Lines 223-281 remain at 12-space indentation directly after the 8-space-indented raise RuntimeError(...) on line 221, with no enclosing block. This is an indentation/parse failure for the whole repository.py module, so dma_api.main (which imports SQLiteMemoryRepository, see services/dma-api/src/dma_api/main.py:69) cannot be imported. Even ignoring parsing, the block is dead code referencing an undefined connection.

Prompt for agents 
In services/dma-api/src/dma_api/repository.py, the refactor of create_or_get into a retry loop (lines 139-221) duplicated the original method body: the old copy still exists at lines 223-281 at an indentation level that no longer belongs to any enclosing block, immediately after the `raise RuntimeError("Unreachable code reached")`. Remove the stale duplicated block so create_or_get contains only the new retry-wrapped implementation, and verify the module imports and the existing test suite passes.
Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Comment thread
krishna3554 marked this conversation as resolved.

def _find_normalized_semantic(self, connection: sqlite3.Connection, record: MemoryRecord) -> MemoryRecord | None:
rows = connection.execute("SELECT * FROM memories WHERE tenant_id = ? AND agent_id = ? AND type = 'semantic'", (record.tenant_id, record.agent_id)).fetchall()
Expand Down
Loading