sources: preserve cursor when inbox persistence fails

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,15 +111,16 @@ 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
         notifications whose thread gains a new comment, or whose ``reason``
         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(

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."""

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"}