From b89480ff1ec34f2fb063d4a3312b78a383255eac Mon Sep 17 00:00:00 2001 From: Oranje AI <293843428+oranjeai@users.noreply.github.com> Date: Tue, 30 Jun 2026 02:12:51 -0100 Subject: [PATCH] Fix source-message updates silently dropped when the updated row is the max rowid insert_source_messages re-surfaces an updated mutable-source record (e.g. a GitHub notification thread that gained a new comment) by deleting the old row and re-inserting it, relying on the re-insert getting a higher rowid so consumer cursors that poll rowid > cursor_seq re-deliver it. source_messages has PRIMARY KEY (source, id) and no AUTOINCREMENT, so the implicit rowid of a re-INSERT is MAX(rowid)+1. When the replaced row is itself the current MAX, deleting it first lowers the max and the re-insert reuses the same rowid, which is <= a consumer cursor already parked there. The update is then never re-delivered. Capture MAX(rowid)+1 before the delete (while the old row still counts toward the max) and re-insert at that explicit rowid, guaranteeing it is strictly above every prior rowid and every consumer cursor. The unchanged-record fast path and the fresh-insert path are untouched. Add tests/test_source_resurface.py covering the max-rowid case (the regression), the non-max case, idempotent no-op re-ingest, and repeated successive updates. Co-Authored-By: Claude Opus 4.8 --- nerve/db/sources.py | 65 +++++++++++----- tests/test_source_resurface.py | 135 +++++++++++++++++++++++++++++++++ 2 files changed, 182 insertions(+), 18 deletions(-) create mode 100644 tests/test_source_resurface.py diff --git a/nerve/db/sources.py b/nerve/db/sources.py index 1d6f4308..ecfb40d9 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -112,10 +112,11 @@ async def insert_source_messages( """Bulk insert source records into the inbox. Returns count inserted. If a record with the same (source, id) already exists but has different - metadata or content, the old record is deleted and re-inserted with a - new rowid. This ensures mutable sources (e.g. GitHub notifications whose - ``reason`` field changes from "author" to "mention") surface as new - messages for consumer cursors. + 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__) @@ -136,32 +137,60 @@ async def insert_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 — delete old record so the - # re-insert gets a new (higher) rowid, making it visible - # to consumer cursors that already read the old version. + # 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 (metadata changed) — re-inserting", - source, r.id, + "Source message %s/%s updated (content/metadata changed): " + "re-inserting at rowid %s to re-surface for consumers", + source, r.id, forced_rowid, ) - 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), - ) + 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) diff --git a/tests/test_source_resurface.py b/tests/test_source_resurface.py new file mode 100644 index 00000000..769154f3 --- /dev/null +++ b/tests/test_source_resurface.py @@ -0,0 +1,135 @@ +"""Re-surfacing of updated mutable-source messages to consumer cursors. + +Mutable sources (notably GitHub) reuse one stable notification id per thread, +so a new comment arrives as an *update* to an existing ``source_messages`` row. +Consumers poll ``rowid > cursor_seq`` and advance their cursor to the max rowid +they have seen, so an updated row must land at a strictly-higher rowid to be +re-delivered. + +Regression guard for the SQLite rowid-reuse bug: ``source_messages`` has +``PRIMARY KEY (source, id)`` and no ``AUTOINCREMENT``, so a naive +delete-then-reinsert lands at ``MAX(rowid)+1``, which *reuses* the old rowid +when the replaced row was itself the table max, leaving the update at or below +a cursor already parked there (silently never re-delivered). +""" + +from __future__ import annotations + +import pytest + +from nerve.sources.models import SourceRecord + + +def _rec(rid: str, content: str, reason: str = "mention") -> SourceRecord: + """A GitHub-ish notification record with a stable per-thread id.""" + return SourceRecord( + id=rid, + source="github", + record_type="github_notification", + summary=f"[owner/repo] thread {rid} ({reason})", + content=content, + timestamp="2026-01-01T00:00:00Z", + metadata={"reason": reason, "repo_name": "owner/repo"}, + ) + + +async def _rowid_of(db, source: str, rid: str) -> int: + async with db.db.execute( + "SELECT rowid FROM source_messages WHERE source = ? AND id = ?", + (source, rid), + ) as cur: + return (await cur.fetchone())[0] + + +@pytest.mark.asyncio +async def test_update_to_max_rowid_row_resurfaces_to_consumer(db): + """The bug's exact shape: the updated thread is the current MAX rowid. + + A consumer reads up to that rowid, then the thread gets a new comment. + The re-inserted row must appear above the consumer's cursor. + """ + # Establish the consumer cursor while the inbox is empty (it initializes to + # the current max rowid = 0), mirroring an inbox cron that has run before. + seq = await db.get_consumer_cursor("inbox", "github") + assert seq == 0 + + # Two earlier threads, then the thread under test; it is now the max rowid. + await db.insert_source_messages([_rec("t1", "a")], source="github") + await db.insert_source_messages([_rec("t2", "b")], source="github") + await db.insert_source_messages([_rec("hot", "comment-1")], source="github") + + # Consumer drains the inbox and parks its cursor at the latest rowid (the + # "hot" thread). This mirrors an inbox cron run that processed the thread. + rows = await db.read_source_messages_by_rowid("github", after_seq=seq, limit=50) + assert {r["id"] for r in rows} == {"t1", "t2", "hot"} + max_seq = max(r["rowid"] for r in rows) + await db.set_consumer_cursor("inbox", "github", max_seq) + parked = await db.get_consumer_cursor("inbox", "github") + + rowid_before = await _rowid_of(db, "github", "hot") + assert rowid_before == parked # the updated thread IS the parked max rowid + + # New comment lands on the same thread -> content changes -> re-insert. + n = await db.insert_source_messages([_rec("hot", "comment-2")], source="github") + assert n == 1 # the changed row was (re-)inserted + + rowid_after = await _rowid_of(db, "github", "hot") + assert rowid_after > parked, ( + f"re-inserted row rowid {rowid_after} must exceed parked cursor {parked} " + "(rowid-reuse regression)" + ) + + # The consumer's next poll re-delivers exactly the updated thread. + fresh = await db.read_source_messages_by_rowid("github", after_seq=parked, limit=50) + assert [r["id"] for r in fresh] == ["hot"] + assert fresh[0]["content"] == "comment-2" + + +@pytest.mark.asyncio +async def test_update_to_non_max_row_also_resurfaces(db): + """The general path: updating a non-max row still re-surfaces above the cursor.""" + base = await db.get_consumer_cursor("inbox", "github") # 0 on empty inbox + await db.insert_source_messages([_rec("old", "x")], source="github") + await db.insert_source_messages([_rec("new", "y")], source="github") + + # Consumer parks at the latest rowid ("new"). + rows = await db.read_source_messages_by_rowid("github", after_seq=base, limit=50) + parked = max(r["rowid"] for r in rows) + await db.set_consumer_cursor("inbox", "github", parked) + + # The OLDER thread (below the cursor) gets new activity. + await db.insert_source_messages([_rec("old", "x-updated")], source="github") + + fresh = await db.read_source_messages_by_rowid("github", after_seq=parked, limit=50) + assert [r["id"] for r in fresh] == ["old"] + assert fresh[0]["content"] == "x-updated" + + +@pytest.mark.asyncio +async def test_unchanged_record_does_not_resurface(db): + """Idempotency preserved: re-ingesting an identical record is a no-op.""" + await db.insert_source_messages([_rec("t", "same")], source="github") + rowid_before = await _rowid_of(db, "github", "t") + + n = await db.insert_source_messages([_rec("t", "same")], source="github") + assert n == 0 # nothing changed -> skipped silently + + rowid_after = await _rowid_of(db, "github", "t") + assert rowid_after == rowid_before # no churn, no spurious re-surface + + +@pytest.mark.asyncio +async def test_repeated_updates_keep_climbing(db): + """Several successive comments on the same thread each re-surface in turn.""" + await db.insert_source_messages([_rec("hot", "c1")], source="github") + last = 0 + for i in range(2, 6): + await db.insert_source_messages([_rec("hot", f"c{i}")], source="github") + rid = await _rowid_of(db, "github", "hot") + assert rid > last, f"rowid must strictly increase on each update (got {rid} <= {last})" + last = rid + # Exactly one live row for the thread (re-insert replaces, not duplicates). + async with db.db.execute( + "SELECT COUNT(*) FROM source_messages WHERE source='github' AND id='hot'" + ) as cur: + assert (await cur.fetchone())[0] == 1