From ab327290cec021fea10a1dd34471f7797211a026 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Tue, 25 Aug 2026 09:51:36 +0200 Subject: [PATCH 1/3] sources: preserve cursor when inbox persistence fails --- nerve/db/sources.py | 134 ++++++++++++------------ nerve/sources/runner.py | 24 +++-- tests/test_source_runner_persistence.py | 96 +++++++++++++++++ 3 files changed, 177 insertions(+), 77 deletions(-) create mode 100644 tests/test_source_runner_persistence.py diff --git a/nerve/db/sources.py b/nerve/db/sources.py index fad21e08..be90500b 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -3,8 +3,11 @@ from __future__ import annotations import json +import logging from datetime import datetime, timedelta, timezone +logger = logging.getLogger(__name__) + class SourceStore: """Mixin providing source inbox, sync cursors, consumer cursors, and run log operations.""" @@ -115,82 +118,77 @@ async def insert_source_messages( field changes from "author" to "mention") surface as new messages for consumer cursors that already read the old version. """ - import logging - logger = logging.getLogger(__name__) now = datetime.now(timezone.utc) expires = (now + timedelta(days=ttl_days)).isoformat() now_iso = now.isoformat() inserted = 0 async with self._atomic(): for r in records: - try: - new_metadata = json.dumps(r.metadata) if r.metadata else None - - # Check if this record already exists + new_metadata = json.dumps(r.metadata) if r.metadata else None + + # Check if this record already exists + async with self.db.execute( + "SELECT metadata, content FROM source_messages " + "WHERE source = ? AND id = ?", + (source, r.id), + ) as cursor: + existing = await cursor.fetchone() + + forced_rowid: int | None = None + if existing: + old_metadata, old_content = existing[0], existing[1] + if old_metadata == new_metadata and old_content == r.content: + # Nothing changed — skip silently + continue + # Metadata or content changed. Re-surface the update to + # consumer cursors (which poll `rowid > cursor_seq`) by + # re-inserting at a strictly-higher rowid. + # + # source_messages has PRIMARY KEY (source, id) and no + # AUTOINCREMENT, so a plain re-INSERT lands at the implicit + # rowid MAX(rowid)+1. If the row being replaced is itself + # the current MAX, deleting it first lowers the max and the + # re-insert REUSES the same rowid, leaving it <= a cursor + # already parked there, so the update is silently never + # re-delivered. Capture MAX(rowid)+1 BEFORE the delete + # (while the old row still counts toward the max) and insert + # at that explicit rowid so it is always above every prior + # rowid and every consumer cursor. async with self.db.execute( - "SELECT metadata, content FROM source_messages " - "WHERE source = ? AND id = ?", - (source, r.id), + "SELECT COALESCE(MAX(rowid), 0) + 1 FROM source_messages" ) as cursor: - existing = await cursor.fetchone() - - forced_rowid: int | None = None - if existing: - old_metadata, old_content = existing[0], existing[1] - if old_metadata == new_metadata and old_content == r.content: - # Nothing changed — skip silently - continue - # Metadata or content changed. Re-surface the update to - # consumer cursors (which poll `rowid > cursor_seq`) by - # re-inserting at a strictly-higher rowid. - # - # source_messages has PRIMARY KEY (source, id) and no - # AUTOINCREMENT, so a plain re-INSERT lands at the implicit - # rowid MAX(rowid)+1. If the row being replaced is itself - # the current MAX, deleting it first lowers the max and the - # re-insert REUSES the same rowid, leaving it <= a cursor - # already parked there, so the update is silently never - # re-delivered. Capture MAX(rowid)+1 BEFORE the delete - # (while the old row still counts toward the max) and insert - # at that explicit rowid so it is always above every prior - # rowid and every consumer cursor. - async with self.db.execute( - "SELECT COALESCE(MAX(rowid), 0) + 1 FROM source_messages" - ) as cursor: - forced_rowid = (await cursor.fetchone())[0] - await self.db.execute( - "DELETE FROM source_messages WHERE source = ? AND id = ?", - (source, r.id), - ) - logger.info( - "Source message %s/%s updated (content/metadata changed): " - "re-inserting at rowid %s to re-surface for consumers", - source, r.id, forced_rowid, - ) - - if forced_rowid is not None: - await self.db.execute( - "INSERT INTO source_messages " - "(rowid, id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - (forced_rowid, r.id, source, r.record_type, r.summary, r.content, - getattr(r, 'raw_content', None), - r.timestamp, new_metadata, - now_iso, expires), - ) - else: - await self.db.execute( - "INSERT INTO source_messages " - "(id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - (r.id, source, r.record_type, r.summary, r.content, - getattr(r, 'raw_content', None), - r.timestamp, new_metadata, - now_iso, expires), - ) - inserted += 1 - except Exception as e: - logger.warning("Failed to insert source message %s: %s", r.id, e) + forced_rowid = (await cursor.fetchone())[0] + await self.db.execute( + "DELETE FROM source_messages WHERE source = ? AND id = ?", + (source, r.id), + ) + logger.info( + "Source message %s/%s updated (content/metadata changed): " + "re-inserting at rowid %s to re-surface for consumers", + source, r.id, forced_rowid, + ) + + if forced_rowid is not None: + await self.db.execute( + "INSERT INTO source_messages " + "(rowid, id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (forced_rowid, r.id, source, r.record_type, r.summary, r.content, + getattr(r, 'raw_content', None), + r.timestamp, new_metadata, + now_iso, expires), + ) + else: + await self.db.execute( + "INSERT INTO source_messages " + "(id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (r.id, source, r.record_type, r.summary, r.content, + getattr(r, 'raw_content', None), + r.timestamp, new_metadata, + now_iso, expires), + ) + inserted += 1 return inserted async def update_source_messages_processed( diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index b122010e..5c3c10f7 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -287,7 +287,18 @@ async def _run_locked(self) -> IngestResult: records = kept # 2. Persist to inbox (post-preprocess, pre-condense — human-readable) - await self._persist_to_inbox(records) + try: + await self._persist_to_inbox(records) + except Exception as e: + logger.error( + "Source %s persistence failed: %s", + self.source.source_name, e, exc_info=True, + ) + return IngestResult( + records_ingested=0, + records_dropped=dropped_count, + error=str(e), + ) # 3. LLM-based condensation for still-long records (configurable per source) if self.condense: @@ -317,14 +328,9 @@ async def _run_locked(self) -> IngestResult: async def _persist_to_inbox(self, records: list[SourceRecord]) -> None: """Save records to the source_messages table for inbox display.""" - try: - await self.db.insert_source_messages( - records, source=self.source.source_name, ttl_days=self.ttl_days, - ) - except Exception as e: - logger.warning( - "Failed to persist %d records to inbox: %s", len(records), e, - ) + await self.db.insert_source_messages( + records, source=self.source.source_name, ttl_days=self.ttl_days, + ) async def _update_processed_content(self, processed_map: dict[str, str]) -> None: """Update processed_content on inbox messages after condensation.""" diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py new file mode 100644 index 00000000..b39e0d02 --- /dev/null +++ b/tests/test_source_runner_persistence.py @@ -0,0 +1,96 @@ +import sqlite3 + +import pytest + +from nerve.sources.base import Source +from nerve.sources.models import FetchResult, SourceRecord +from nerve.sources.runner import SourceRunner + + +SOURCE_NAME = "test-source-persistence" + + +def _record(record_id: str) -> SourceRecord: + return SourceRecord( + id=record_id, + source=SOURCE_NAME, + record_type="test", + summary=f"Record {record_id}", + content=f"Content {record_id}", + timestamp="2026-08-25T00:00:00Z", + ) + + +class _FixedSource(Source): + source_name = SOURCE_NAME + + def __init__(self, records: list[SourceRecord], next_cursor: str = "new"): + self.records = records + self.next_cursor = next_cursor + + async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: + return FetchResult(records=list(self.records), next_cursor=self.next_cursor) + + +@pytest.mark.asyncio +async def test_insert_source_messages_rolls_back_the_batch_on_insert_failure(db): + records = [_record("a"), _record("b"), _record("c")] + records[1].summary = None + + with pytest.raises(sqlite3.IntegrityError): + await db.insert_source_messages(records, source=SOURCE_NAME) + + rows, _ = await db.list_source_messages(source=SOURCE_NAME, limit=10) + assert rows == [] + assert not db.db.in_transaction + + +@pytest.mark.asyncio +async def test_runner_reports_persistence_failure_without_advancing_cursor( + db, monkeypatch, +): + await db.set_sync_cursor(SOURCE_NAME, "old") + runner = SourceRunner(_FixedSource([_record("a")]), db) + + async def fail(records, source, ttl_days): + raise RuntimeError("inbox unavailable") + + monkeypatch.setattr(db, "insert_source_messages", fail) + + result = await runner.run() + + assert result.records_ingested == 0 + assert result.error == "inbox unavailable" + assert await db.get_sync_cursor(SOURCE_NAME) == "old" + + +@pytest.mark.asyncio +async def test_runner_retries_the_same_batch_after_persistence_recovers( + db, monkeypatch, +): + records = [_record("a"), _record("b")] + runner = SourceRunner(_FixedSource(records), db) + insert_source_messages = db.insert_source_messages + attempts = 0 + + async def fail_once(records, source, ttl_days): + nonlocal attempts + attempts += 1 + if attempts == 1: + raise RuntimeError("temporary inbox failure") + return await insert_source_messages(records, source=source, ttl_days=ttl_days) + + monkeypatch.setattr(db, "insert_source_messages", fail_once) + + first = await runner.run() + assert first.error == "temporary inbox failure" + assert await db.get_sync_cursor(SOURCE_NAME) is None + + runner.health.backoff_until = None + second = await runner.run() + + assert second.error is None + assert second.records_ingested == 2 + assert await db.get_sync_cursor(SOURCE_NAME) == "new" + rows, _ = await db.list_source_messages(source=SOURCE_NAME, limit=10) + assert {row["id"] for row in rows} == {"a", "b"} From 7942d5e3d136fa06313c2436f8178206864b8d6e Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Tue, 25 Aug 2026 10:02:13 +0200 Subject: [PATCH 2/3] sources: include failing record in persistence errors --- nerve/db/sources.py | 134 +++++++++++++----------- tests/test_source_runner_persistence.py | 20 ++++ 2 files changed, 92 insertions(+), 62 deletions(-) diff --git a/nerve/db/sources.py b/nerve/db/sources.py index be90500b..5db130f9 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -111,6 +111,9 @@ async def insert_source_messages( ) -> int: """Bulk insert source records into the inbox. Returns count inserted. + The batch is atomic; if any record cannot be persisted, no records are + committed and the exception is propagated. + If a record with the same (source, id) already exists but has different metadata or content, the old record is deleted and re-inserted at a strictly-higher rowid. This ensures mutable sources (e.g. GitHub @@ -124,71 +127,78 @@ async def insert_source_messages( inserted = 0 async with self._atomic(): for r in records: - new_metadata = json.dumps(r.metadata) if r.metadata else None - - # Check if this record already exists - async with self.db.execute( - "SELECT metadata, content FROM source_messages " - "WHERE source = ? AND id = ?", - (source, r.id), - ) as cursor: - existing = await cursor.fetchone() - - forced_rowid: int | None = None - if existing: - old_metadata, old_content = existing[0], existing[1] - if old_metadata == new_metadata and old_content == r.content: - # Nothing changed — skip silently - continue - # Metadata or content changed. Re-surface the update to - # consumer cursors (which poll `rowid > cursor_seq`) by - # re-inserting at a strictly-higher rowid. - # - # source_messages has PRIMARY KEY (source, id) and no - # AUTOINCREMENT, so a plain re-INSERT lands at the implicit - # rowid MAX(rowid)+1. If the row being replaced is itself - # the current MAX, deleting it first lowers the max and the - # re-insert REUSES the same rowid, leaving it <= a cursor - # already parked there, so the update is silently never - # re-delivered. Capture MAX(rowid)+1 BEFORE the delete - # (while the old row still counts toward the max) and insert - # at that explicit rowid so it is always above every prior - # rowid and every consumer cursor. + try: + new_metadata = json.dumps(r.metadata) if r.metadata else None + + # Check if this record already exists async with self.db.execute( - "SELECT COALESCE(MAX(rowid), 0) + 1 FROM source_messages" - ) as cursor: - forced_rowid = (await cursor.fetchone())[0] - await self.db.execute( - "DELETE FROM source_messages WHERE source = ? AND id = ?", + "SELECT metadata, content FROM source_messages " + "WHERE source = ? AND id = ?", (source, r.id), + ) as cursor: + existing = await cursor.fetchone() + + forced_rowid: int | None = None + if existing: + old_metadata, old_content = existing[0], existing[1] + if old_metadata == new_metadata and old_content == r.content: + # Nothing changed — skip silently + continue + # Metadata or content changed. Re-surface the update to + # consumer cursors (which poll `rowid > cursor_seq`) by + # re-inserting at a strictly-higher rowid. + # + # source_messages has PRIMARY KEY (source, id) and no + # AUTOINCREMENT, so a plain re-INSERT lands at the implicit + # rowid MAX(rowid)+1. If the row being replaced is itself + # the current MAX, deleting it first lowers the max and the + # re-insert REUSES the same rowid, leaving it <= a cursor + # already parked there, so the update is silently never + # re-delivered. Capture MAX(rowid)+1 BEFORE the delete + # (while the old row still counts toward the max) and insert + # at that explicit rowid so it is always above every prior + # rowid and every consumer cursor. + async with self.db.execute( + "SELECT COALESCE(MAX(rowid), 0) + 1 FROM source_messages" + ) as cursor: + forced_rowid = (await cursor.fetchone())[0] + await self.db.execute( + "DELETE FROM source_messages WHERE source = ? AND id = ?", + (source, r.id), + ) + logger.info( + "Source message %s/%s updated (content/metadata changed): " + "re-inserting at rowid %s to re-surface for consumers", + source, r.id, forced_rowid, + ) + + if forced_rowid is not None: + await self.db.execute( + "INSERT INTO source_messages " + "(rowid, id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (forced_rowid, r.id, source, r.record_type, r.summary, r.content, + getattr(r, 'raw_content', None), + r.timestamp, new_metadata, + now_iso, expires), + ) + else: + await self.db.execute( + "INSERT INTO source_messages " + "(id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", + (r.id, source, r.record_type, r.summary, r.content, + getattr(r, 'raw_content', None), + r.timestamp, new_metadata, + now_iso, expires), + ) + inserted += 1 + except Exception as e: + logger.warning( + "Failed to persist source message %s/%s: %s", + source, r.id, e, ) - logger.info( - "Source message %s/%s updated (content/metadata changed): " - "re-inserting at rowid %s to re-surface for consumers", - source, r.id, forced_rowid, - ) - - if forced_rowid is not None: - await self.db.execute( - "INSERT INTO source_messages " - "(rowid, id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - (forced_rowid, r.id, source, r.record_type, r.summary, r.content, - getattr(r, 'raw_content', None), - r.timestamp, new_metadata, - now_iso, expires), - ) - else: - await self.db.execute( - "INSERT INTO source_messages " - "(id, source, record_type, summary, content, raw_content, timestamp, metadata, created_at, expires_at) " - "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", - (r.id, source, r.record_type, r.summary, r.content, - getattr(r, 'raw_content', None), - r.timestamp, new_metadata, - now_iso, expires), - ) - inserted += 1 + raise return inserted async def update_source_messages_processed( diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index b39e0d02..0e8be7d7 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -64,6 +64,26 @@ async def fail(records, source, ttl_days): assert await db.get_sync_cursor(SOURCE_NAME) == "old" +@pytest.mark.asyncio +async def test_runner_rolls_back_real_insert_failure_and_preserves_cursor( + db, caplog, +): + await db.set_sync_cursor(SOURCE_NAME, "old") + records = [_record("good"), _record("bad")] + records[1].summary = None + runner = SourceRunner(_FixedSource(records), db) + + result = await runner.run() + + assert result.records_ingested == 0 + assert result.error is not None + assert await db.get_sync_cursor(SOURCE_NAME) == "old" + rows, _ = await db.list_source_messages(source=SOURCE_NAME, limit=10) + assert rows == [] + assert not db.db.in_transaction + assert f"{SOURCE_NAME}/bad" in caplog.text + + @pytest.mark.asyncio async def test_runner_retries_the_same_batch_after_persistence_recovers( db, monkeypatch, From 2fefe259bc879cb1a967d325b0b7a8f96c849d09 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Tue, 25 Aug 2026 10:16:39 +0200 Subject: [PATCH 3/3] test: cover rollback of mutable source updates --- tests/test_source_runner_persistence.py | 37 +++++++++++++++++++++++++ 1 file changed, 37 insertions(+) diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index 0e8be7d7..6f523b04 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -84,6 +84,43 @@ async def test_runner_rolls_back_real_insert_failure_and_preserves_cursor( assert f"{SOURCE_NAME}/bad" in caplog.text +@pytest.mark.asyncio +async def test_runner_rolls_back_existing_update_when_later_insert_fails(db): + existing = _record("existing") + await db.insert_source_messages([existing], source=SOURCE_NAME) + + async with db.db.execute( + "SELECT rowid FROM source_messages WHERE source = ? AND id = ?", + (SOURCE_NAME, existing.id), + ) as cursor: + old_rowid = (await cursor.fetchone())[0] + + await db.set_sync_cursor(SOURCE_NAME, "old") + updated = _record(existing.id) + updated.content = "Updated content" + bad = _record("bad") + bad.summary = None + + result = await SourceRunner(_FixedSource([updated, bad]), db).run() + + assert result.records_ingested == 0 + assert result.error is not None + assert await db.get_sync_cursor(SOURCE_NAME) == "old" + async with db.db.execute( + "SELECT rowid, id, summary, content FROM source_messages " + "WHERE source = ? ORDER BY rowid", + (SOURCE_NAME,), + ) as cursor: + rows = [dict(row) async for row in cursor] + assert rows == [{ + "rowid": old_rowid, + "id": existing.id, + "summary": existing.summary, + "content": existing.content, + }] + assert not db.db.in_transaction + + @pytest.mark.asyncio async def test_runner_retries_the_same_batch_after_persistence_recovers( db, monkeypatch,