diff --git a/nerve/db/sources.py b/nerve/db/sources.py index fad21e08..5db130f9 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.""" @@ -108,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 @@ -115,8 +121,6 @@ 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() @@ -190,7 +194,11 @@ async def insert_source_messages( ) inserted += 1 except Exception as e: - logger.warning("Failed to insert source message %s: %s", r.id, e) + logger.warning( + "Failed to persist source message %s/%s: %s", + source, r.id, e, + ) + raise 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..6f523b04 --- /dev/null +++ b/tests/test_source_runner_persistence.py @@ -0,0 +1,153 @@ +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_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_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, +): + 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"}