From 6262c28a4834f07bac14143e3347d7b2cdc43937 Mon Sep 17 00:00:00 2001 From: Abhay Singh Date: Tue, 25 Aug 2026 11:40:54 +0530 Subject: [PATCH] fix(memory/graph): skip a corrupt row instead of aborting a whole graph scan (#3239) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description `SQLiteGraphStore._row_to_entity` and `_row_to_relationship` parse stored text back into objects with no error handling: ```python properties=json.loads(row["properties"]), created_at=datetime.fromisoformat(row["created_at"]), metadata=json.loads(row["metadata"]), ``` These run inside row loops in the multi-row scans — `get_relationships` and `query_subgraph` (both the relationship loop and neighbour-entity expansion). A single unparseable row — from a partial write, a manual edit, or a bad migration — raises `ValueError` (`JSONDecodeError`/bad ISO timestamp) *inside the loop*, aborting the **entire** query and taking unrelated, perfectly good edges/nodes down with it. Reproduction (A→B and A→C both valid; corrupt only A→B's `properties`): ```python # corrupt one row out-of-band con.execute("UPDATE relationships SET properties='{oops' WHERE target_id=?", (b.id,)) # BEFORE: both of these raise JSONDecodeError, even though A->C is fine: await store.get_relationships(a.id) await store.query_subgraph([a.id], max_hops=1, direction=OUTGOING) ``` This is the same "one bad row breaks the whole scan" robustness gap already fixed for the CCR store (`cache/backends/sqlite.py`) and the vector adapter (`memory/adapters/sqlite_vector.py`); the graph adapter was the remaining store with unguarded row parsing. ## Type of Change - [x] Bug fix (non-breaking change that fixes an issue) - [ ] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `headroom/memory/adapters/sqlite_graph.py`: - `_row_to_entity` / `_row_to_relationship` now return `... | None`, wrapping construction in `except (ValueError, TypeError, KeyError)` and returning `None` (with a `logger.warning`) on a corrupt row. - Multi-row call sites skip `None`: `get_relationships`, `query_subgraph` (initial entities, relationship loop, neighbour expansion), and the per-user entity listing. The single-row `get_entity` / `get_entity_by_name` already return `Entity | None`, so a corrupt row now reads as "not found" rather than raising. - Added a module `logger`. - `tests/test_sqlite_graph_store.py`: added `test_one_corrupt_row_does_not_abort_a_multi_row_scan` — corrupts one relationship row out-of-band and asserts `get_relationships` returns the one good edge and `query_subgraph` completes with `{A, C}`. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality ### Test Output ```text tests/test_sqlite_graph_store.py::...one_corrupt_row_does_not_abort_a_multi_row_scan -> passes with fix, FAILS without it (verified via git stash) uvx ruff@0.16.2 check headroom/memory/adapters/sqlite_graph.py tests/test_sqlite_graph_store.py -> All checks passed! uvx mypy@1.20.2 headroom/memory/adapters/sqlite_graph.py -> Success: no issues found in 1 source file ``` (Note: this test file has pre-existing, unrelated failures/errors on `main` on Windows — `TestSQLiteGraphStoreMemoryTrackerIntegration` plus temp-file teardown `WinError 32` in the `NamedTemporaryFile`-based fixtures. Verified identical counts before and after this change; my new test uses `tmp_path` and is unaffected.) ## Real Behavior Proof - Environment: Windows 11, Python 3.12.11, project venv, pytest 9.1.1, ruff 0.16.2 and mypy 1.20.2 via uvx. - Exact command / steps: built A→B and A→C edges, corrupted A→B's `properties` to invalid JSON via a direct sqlite connection, then called `get_relationships(A)` and `query_subgraph([A], OUTGOING)`. Before the fix both raised `JSONDecodeError`; after the fix `get_relationships` returns just the A→C edge and `query_subgraph` returns entities `{A, C}` with one relationship, skipping the corrupt row. - Observed result: corrupt rows are skipped (with a warning log); valid rows in the same scan are returned normally. - Not tested: no corruption occurs in normal operation; the corrupt row is produced out-of-band to exercise the guard (matching the real triggers: partial write, manual edit, migration). ## Runtime Rollout Safety - Rollout-managed feature(s): none. This is a SQLite graph-store read path in the memory subsystem, not a rollout-channel-gated runtime feature. - Minimum rollout channel: N/A. - Stable/default behavior changed: no for well-formed data — every valid row parses and is returned exactly as before. Only the previously-crashing corrupt-row case changes, from an aborted query to a skipped row. - Kill switch / disable path: N/A. - Unsafe override required: no. - Qualification impact: none. - Rollback path: revert this PR. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation (N/A: internal behavior) - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] I did **not** edit `CHANGELOG.md` --- headroom/memory/adapters/sqlite_graph.py | 97 ++++++++++++++++-------- tests/test_sqlite_graph_store.py | 43 +++++++++++ 2 files changed, 110 insertions(+), 30 deletions(-) diff --git a/headroom/memory/adapters/sqlite_graph.py b/headroom/memory/adapters/sqlite_graph.py index 2c2ac432c..8c74bde6f 100644 --- a/headroom/memory/adapters/sqlite_graph.py +++ b/headroom/memory/adapters/sqlite_graph.py @@ -14,6 +14,7 @@ This is a drop-in replacement for InMemoryGraphStore that: from __future__ import annotations import json +import logging import sqlite3 from collections import deque from datetime import datetime @@ -26,6 +27,8 @@ from .graph_models import Entity, Relationship, RelationshipDirection, Subgraph if TYPE_CHECKING: from ..tracker import ComponentStats +logger = logging.getLogger(__name__) + class SQLiteGraphStore: """SQLite-based graph store implementing the GraphStore protocol. @@ -165,19 +168,31 @@ class SQLiteGraphStore: "metadata": json.dumps(entity.metadata), } - def _row_to_entity(self, row: sqlite3.Row) -> Entity: - """Convert database row to Entity object.""" - return Entity( - id=row["id"], - user_id=row["user_id"], - name=row["name"], - entity_type=row["entity_type"], - description=row["description"], - properties=json.loads(row["properties"]), - created_at=datetime.fromisoformat(row["created_at"]), - updated_at=datetime.fromisoformat(row["updated_at"]), - metadata=json.loads(row["metadata"]), - ) + def _row_to_entity(self, row: sqlite3.Row) -> Entity | None: + """Convert a database row to an Entity, or None if the row is corrupt. + + ``properties``/``metadata`` (JSON) and ``created_at``/``updated_at`` + (ISO timestamps) are parsed from stored text. A single unparseable row — + from a partial write, a manual edit, or a bad migration — must not abort + an entire multi-row scan (``query_subgraph``, neighbour expansion): one + corrupt edge would otherwise make an unrelated part of the graph + unqueryable. Skip the bad row instead. + """ + try: + return Entity( + id=row["id"], + user_id=row["user_id"], + name=row["name"], + entity_type=row["entity_type"], + description=row["description"], + properties=json.loads(row["properties"]), + created_at=datetime.fromisoformat(row["created_at"]), + updated_at=datetime.fromisoformat(row["updated_at"]), + metadata=json.loads(row["metadata"]), + ) + except (ValueError, TypeError, KeyError) as exc: + logger.warning("skipping corrupt entity row %r: %s", row["id"], exc) + return None def _relationship_to_row(self, relationship: Relationship) -> dict[str, Any]: """Convert Relationship object to row dict for insertion.""" @@ -193,19 +208,29 @@ class SQLiteGraphStore: "metadata": json.dumps(relationship.metadata), } - def _row_to_relationship(self, row: sqlite3.Row) -> Relationship: - """Convert database row to Relationship object.""" - return Relationship( - id=row["id"], - user_id=row["user_id"], - source_id=row["source_id"], - target_id=row["target_id"], - relation_type=row["relation_type"], - weight=row["weight"], - properties=json.loads(row["properties"]), - created_at=datetime.fromisoformat(row["created_at"]), - metadata=json.loads(row["metadata"]), - ) + def _row_to_relationship(self, row: sqlite3.Row) -> Relationship | None: + """Convert a database row to a Relationship, or None if the row is corrupt. + + Same contract as :meth:`_row_to_entity`: a single unparseable relationship + row (bad ``properties``/``metadata`` JSON or ``created_at`` timestamp) must + not abort a whole ``get_relationships`` / ``query_subgraph`` scan and take + unrelated edges down with it. Skip the bad row instead. + """ + try: + return Relationship( + id=row["id"], + user_id=row["user_id"], + source_id=row["source_id"], + target_id=row["target_id"], + relation_type=row["relation_type"], + weight=row["weight"], + properties=json.loads(row["properties"]), + created_at=datetime.fromisoformat(row["created_at"]), + metadata=json.loads(row["metadata"]), + ) + except (ValueError, TypeError, KeyError) as exc: + logger.warning("skipping corrupt relationship row %r: %s", row["id"], exc) + return None # ========================================================================= # Entity Operations @@ -373,7 +398,9 @@ class SQLiteGraphStore: params, ) - return [self._row_to_relationship(row) for row in cursor] + return [ + rel for row in cursor if (rel := self._row_to_relationship(row)) is not None + ] async def delete_relationship(self, relationship_id: str) -> bool: """Delete a single relationship. @@ -436,9 +463,12 @@ class SQLiteGraphStore: ) row = cursor.fetchone() if row is not None: + entity = self._row_to_entity(row) + if entity is None: + continue queue.append((entity_id, 0)) visited.add(entity_id) - collected_entities[entity_id] = self._row_to_entity(row) + collected_entities[entity_id] = entity # BFS traversal while queue: @@ -470,6 +500,8 @@ class SQLiteGraphStore: for rel_row in cursor: rel = self._row_to_relationship(rel_row) + if rel is None: + continue # Add relationship collected_relationships[rel.id] = rel @@ -496,8 +528,11 @@ class SQLiteGraphStore: ) neighbor_row = neighbor_cursor.fetchone() if neighbor_row is not None: + neighbor = self._row_to_entity(neighbor_row) + if neighbor is None: + continue visited.add(neighbor_id) - collected_entities[neighbor_id] = self._row_to_entity(neighbor_row) + collected_entities[neighbor_id] = neighbor queue.append((neighbor_id, depth + 1)) return Subgraph( @@ -651,7 +686,9 @@ class SQLiteGraphStore: "SELECT * FROM entities WHERE user_id = ?", (user_id,), ) - return [self._row_to_entity(row) for row in cursor] + return [ + entity for row in cursor if (entity := self._row_to_entity(row)) is not None + ] async def clear(self) -> None: """Clear all data from the store.""" diff --git a/tests/test_sqlite_graph_store.py b/tests/test_sqlite_graph_store.py index 9fb5a90b0..b3db37e62 100644 --- a/tests/test_sqlite_graph_store.py +++ b/tests/test_sqlite_graph_store.py @@ -689,6 +689,49 @@ class TestSQLiteGraphStoreEdgeCases: assert len(subgraph.entities) == 0 assert len(subgraph.relationships) == 0 + @pytest.mark.asyncio + async def test_one_corrupt_row_does_not_abort_a_multi_row_scan(self, tmp_path): + """A single unparseable row must not take down an entire query. + + Regression: ``_row_to_relationship`` / ``_row_to_entity`` parsed stored + JSON/timestamps with no guard, so one corrupt row (partial write, manual + edit, bad migration) raised inside the row loop and aborted the whole + ``query_subgraph`` / ``get_relationships`` scan — taking unrelated, + perfectly good edges down with it. The corrupt row is now skipped. + """ + import sqlite3 + + store = SQLiteGraphStore(db_path=str(tmp_path / "graph.db")) + a = Entity(user_id="u", name="A", entity_type="n") + b = Entity(user_id="u", name="B", entity_type="n") + c = Entity(user_id="u", name="C", entity_type="n") + for entity in (a, b, c): + await store.add_entity(entity) + await store.add_relationship( + Relationship(user_id="u", source_id=a.id, target_id=b.id, relation_type="e") + ) + await store.add_relationship( + Relationship(user_id="u", source_id=a.id, target_id=c.id, relation_type="e") + ) + + # Corrupt the A->B relationship row's properties JSON out-of-band. + con = sqlite3.connect(str(store.db_path)) + con.execute("UPDATE relationships SET properties = '{oops' WHERE target_id = ?", (b.id,)) + con.commit() + con.close() + + # get_relationships returns the one good edge instead of raising. + rels = await store.get_relationships(a.id) + assert len(rels) == 1 + assert rels[0].target_id == c.id + + # query_subgraph completes, skipping the corrupt edge and its node. + subgraph = await store.query_subgraph( + [a.id], max_hops=1, direction=RelationshipDirection.OUTGOING + ) + assert {e.name for e in subgraph.entities} == {"A", "C"} + assert len(subgraph.relationships) == 1 + @pytest.mark.asyncio async def test_entity_with_special_characters(self, store): """Test entity names with special characters."""