From eb7f6a0659726573db19482eaef9ceb2213fa7f0 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 07:22:10 +0000 Subject: [PATCH 1/4] Sources: start consumer cursors before a source's first run A consumer that reads other sources does not see the first records of a new source. consumer_has_unread ignores a source without a cursor for the consumer, so a messages gate on any source does not fire for it. The consumer's first poll then starts its cursor at the newest message, past the first records. This affects github_repos and every source that starts later. Before a source's first run (no source cursor), start a cursor on the source for every consumer that has a cursor on another source, at the source's newest message. A failure is logged and the run continues. Based on a fix by Frank Rosner. Co-authored-by: Frank Rosner <3427394+FRosner@users.noreply.github.com> Co-Authored-By: Claude Opus 5.5 --- docs/sources.md | 1 + nerve/db/sources.py | 23 ++++++++ nerve/sources/runner.py | 27 +++++++++ tests/test_hosted_sources.py | 17 +++++- tests/test_source_runner_persistence.py | 77 +++++++++++++++++++++++++ 5 files changed, 144 insertions(+), 1 deletion(-) diff --git a/docs/sources.md b/docs/sources.md index 6b50d6083..9b09b94ff 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -55,6 +55,7 @@ Consumers read from the inbox using the `poll_source` agent tool with a named co **Key properties:** - Consumer cursors are independent — multiple consumers can read the same source at different positions - New consumer cursors initialize to the latest message (no backlog flooding) +- Before a source's first run, each consumer that has a cursor on another source gets a cursor on the new source, so the first records of the new source are unread for that consumer and its `messages` gates see them - Consumer cursors expire after N days of inactivity (configurable, default: 2 days); a `messages` gate that checks the consumer renews the cursors it reads (see [cron.md](cron.md#messages)) - Cross-source deduplication happens naturally because one agent session sees messages from all sources diff --git a/nerve/db/sources.py b/nerve/db/sources.py index 66cdaab95..ccfdfcb66 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -399,6 +399,29 @@ async def set_consumer_cursor( (consumer, source, cursor_seq, session_id, now.isoformat(), expires), ) + async def seed_consumer_cursors(self, source: str, ttl_days: int = 2) -> int: + """Start a cursor on *source* for every consumer that has one elsewhere. + + Each starts at the source's newest message, as :meth:`get_consumer_cursor` + would, so what the source stores after this call is unread. Without it + a new source stays invisible to :meth:`consumer_has_unread`, and the + consumer's first poll starts past the source's first records. A + consumer's existing cursor on *source* is left as it is. Returns the + number of cursors started. + """ + now = datetime.now(UTC) + expires = (now + timedelta(days=ttl_days)).isoformat() + result = await self._write( + """INSERT OR IGNORE INTO consumer_cursors + (consumer, source, cursor_seq, updated_at, expires_at) + SELECT DISTINCT consumer, ?, + (SELECT COALESCE(MAX(rowid), 0) FROM source_messages WHERE source = ?), + ?, ? + FROM consumer_cursors WHERE source != ?""", + (source, source, now.isoformat(), expires, source), + ) + return result.rowcount or 0 + async def keep_consumer_cursors_alive( self, consumer: str, sources: list[str] | None = None, ttl_days: int = 2, ) -> int: diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index 39cd338ea..685f14453 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -322,6 +322,8 @@ async def _run_locked(self) -> IngestResult: there. """ cursor = await self.db.get_sync_cursor(self.source.source_name) + if cursor is None: + await self._seed_consumer_cursors() total_ingested = 0 total_dropped = 0 @@ -361,6 +363,31 @@ async def _run_locked(self) -> IngestResult: records_ingested=total_ingested, records_dropped=total_dropped, ) + async def _seed_consumer_cursors(self) -> None: + """Before a source's first run, start the existing consumers' cursors on it. + + This occurs before the source stores a record, so its first records + are unread for those consumers, and their ``messages`` gates see them. + Best effort: a failure is logged and the run continues. + """ + from nerve.cron.gates import _consumer_cursor_ttl_days + + try: + seeded = await self.db.seed_consumer_cursors( + self.source.source_name, ttl_days=_consumer_cursor_ttl_days(), + ) + except Exception: # noqa: BLE001 + logger.warning( + "Source %s: could not start the consumers' cursors", + self.source.source_name, exc_info=True, + ) + return + if seeded: + logger.info( + "Source %s: first run, started %d consumer cursor(s) on it", + self.source.source_name, seeded, + ) + async def _ingest_batch(self, cursor: str | None) -> _BatchResult: """One fetch → preprocess → guardrail → persist → condense → drop hook → advance cycle.""" logger.info( diff --git a/tests/test_hosted_sources.py b/tests/test_hosted_sources.py index ef0de0949..5d2214581 100644 --- a/tests/test_hosted_sources.py +++ b/tests/test_hosted_sources.py @@ -59,12 +59,13 @@ SlackConfig, validate_config_keys, ) -from nerve.cron.gates import GateContext, GitHubPrActivityGate +from nerve.cron.gates import GateContext, GitHubPrActivityGate, MessagesGate from nerve.cron.service import CronService from nerve.gateway import server as gateway_server from nerve.sources.github import GitHubSource, notification_record from nerve.sources.github_hosted import HostedGitHubSource, record_from_buffer from nerve.sources.registry import build_source_runners +from nerve.sources.runner import SourceRunner from tests.fake_channel_gateway import ( EVENT_TIMEOUT, FakeChannelGateway, @@ -652,6 +653,20 @@ async def test_the_drain_reads_past_a_polling_cursor_and_skips_unreadable_rows(s again = await HostedGitHubSource(db).fetch(result.next_cursor, limit=10) assert (again.records, again.next_cursor) == ([], "2") + @pytest.mark.asyncio + async def test_the_first_drain_of_a_provider_wakes_the_consumers_of_other_sources(self, db): + """The first records of a provider that starts later, here the + repository watch, are unread for a consumer that reads another source.""" + await db.set_consumer_cursor("inbox", "github", 0) + await db.insert_channel_observation("github_repos", "github_repos:owner/repo", self.payload()) + gate, ctx = MessagesGate(), GateContext(job_id="inbox-processor", db=db) + assert not await gate.is_satisfied(ctx) + + await SourceRunner(HostedGitHubSource(db, provider="github_repos"), db).run() + + assert await gate.is_satisfied(ctx) + assert await db.peek_consumer_cursor("inbox", "github_repos") == 0 + @pytest.mark.asyncio class TestRegistry: diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index 18852a765..410fd0811 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -3,6 +3,9 @@ import pytest +from nerve.agent.tools.handlers.sources import poll_all_sources_handler +from nerve.agent.tools.registry import ToolContext +from nerve.cron.gates import GateContext, MessagesGate from nerve.sources.base import Source from nerve.sources.models import FetchResult, SourceRecord from nerve.sources.runner import SourceRunner @@ -192,3 +195,77 @@ async def fake_condense(records): assert result.records_ingested == 1 assert condensed == [[repeat_content]] + + +@pytest.mark.asyncio +async def test_a_new_source_s_first_records_are_unread_for_the_existing_consumers(db): + # The inbox consumer already reads another source. + await db.insert_source_messages([dataclasses.replace(_record("x"), source="other")], source="other") + await db.set_consumer_cursor("inbox", "other", await db.get_source_max_rowid("other")) + ctx = GateContext(job_id="inbox-processor", db=db) + assert not await MessagesGate().is_satisfied(ctx) + + result = await SourceRunner(_FixedSource([_record("a")]), db).run() + + assert result.records_ingested == 1 + assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 + # A gate on any source and a gate on the new source both see the records. + assert await MessagesGate().is_satisfied(ctx) + assert await MessagesGate(sources=[SOURCE_NAME]).is_satisfied(ctx) + messages = await db.read_source_messages_by_rowid( + SOURCE_NAME, after_seq=await db.get_consumer_cursor("inbox", SOURCE_NAME), + ) + assert [m["id"] for m in messages] == ["a"] + + +@pytest.mark.asyncio +async def test_a_baseline_run_seeds_the_cursor_so_later_records_wake_the_consumer(db): + await db.set_consumer_cursor("inbox", "other", 0) + source = _FixedSource([], next_cursor="baseline") + runner = SourceRunner(source, db) + + await runner.run() + assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 + assert not await db.consumer_has_unread("inbox") + + source.records, source.next_cursor = [_record("later")], "next" + await runner.run() + assert await db.consumer_has_unread("inbox") + + text = (await poll_all_sources_handler( + ToolContext(session_id="s", db=db), {"consumer": "inbox"}, + )).text_content + assert "Record later" in text + assert f"{SOURCE_NAME}: 1 new" in text + + +@pytest.mark.asyncio +async def test_seeding_leaves_an_existing_cursor_and_only_runs_without_a_source_cursor(db): + await db.insert_source_messages([_record("old")], source=SOURCE_NAME) + await db.set_consumer_cursor("inbox", SOURCE_NAME, 0) + await db.set_consumer_cursor("other-consumer", "other", 0) + + assert await db.seed_consumer_cursors(SOURCE_NAME) == 1 + assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 + # A consumer new to the source starts past what is already stored. + assert await db.peek_consumer_cursor("other-consumer", SOURCE_NAME) == ( + await db.get_source_max_rowid(SOURCE_NAME) + ) + + await db.set_sync_cursor(SOURCE_NAME, "stored") + await db.set_consumer_cursor("late", "other", 0) + await SourceRunner(_FixedSource([]), db).run() + assert await db.peek_consumer_cursor("late", SOURCE_NAME) is None + + +@pytest.mark.asyncio +async def test_a_failure_to_start_the_consumer_cursors_does_not_stop_the_run(db, monkeypatch): + async def fail(source, ttl_days): + raise RuntimeError("cursors unavailable") + + monkeypatch.setattr(db, "seed_consumer_cursors", fail) + + result = await SourceRunner(_FixedSource([_record("a")]), db).run() + + assert (result.records_ingested, result.error) == (1, None) + assert await db.get_sync_cursor(SOURCE_NAME) == "new" From 4f73f7d3567db5222a081a77d91b909258ae53f8 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 08:31:31 +0000 Subject: [PATCH 2/4] Sources: fail the run when the consumer cursors cannot start The runner started the consumer cursors on a new source as a best-effort step: on a failure it logged a warning and the run went on. The run then stored the first records and set the source cursor, so the step did not run again, and the existing consumers did not get the first records. Stop the run before the fetch when the step fails, with the error in the result, as for a fetch failure. The source then stores nothing and has no cursor, so the next run tries again. Co-Authored-By: Claude Opus 5.5 --- nerve/sources/runner.py | 21 ++++++++----- tests/test_source_runner_persistence.py | 39 ++++++++++++++++++++----- 2 files changed, 44 insertions(+), 16 deletions(-) diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index 685f14453..e2b151433 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -323,7 +323,9 @@ async def _run_locked(self) -> IngestResult: """ cursor = await self.db.get_sync_cursor(self.source.source_name) if cursor is None: - await self._seed_consumer_cursors() + error = await self._seed_consumer_cursors() + if error: + return IngestResult(records_ingested=0, error=error) total_ingested = 0 total_dropped = 0 @@ -363,12 +365,14 @@ async def _run_locked(self) -> IngestResult: records_ingested=total_ingested, records_dropped=total_dropped, ) - async def _seed_consumer_cursors(self) -> None: + async def _seed_consumer_cursors(self) -> str | None: """Before a source's first run, start the existing consumers' cursors on it. This occurs before the source stores a record, so its first records are unread for those consumers, and their ``messages`` gates see them. - Best effort: a failure is logged and the run continues. + Returns the error text of a failure, else ``None``. On a failure the + run stops before the fetch. The source then stores nothing and still + has no cursor, so the next run tries again. """ from nerve.cron.gates import _consumer_cursor_ttl_days @@ -376,17 +380,18 @@ async def _seed_consumer_cursors(self) -> None: seeded = await self.db.seed_consumer_cursors( self.source.source_name, ttl_days=_consumer_cursor_ttl_days(), ) - except Exception: # noqa: BLE001 - logger.warning( - "Source %s: could not start the consumers' cursors", - self.source.source_name, exc_info=True, + except Exception as e: # noqa: BLE001 + logger.error( + "Source %s: could not start the consumers' cursors: %s", + self.source.source_name, e, exc_info=True, ) - return + return f"could not start the consumer cursors: {e}" if seeded: logger.info( "Source %s: first run, started %d consumer cursor(s) on it", self.source.source_name, seeded, ) + return None async def _ingest_batch(self, cursor: str | None) -> _BatchResult: """One fetch → preprocess → guardrail → persist → condense → drop hook → advance cycle.""" diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index 410fd0811..e27e416d3 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -31,11 +31,23 @@ class _FixedSource(Source): def __init__(self, records: list[SourceRecord], next_cursor: str = "new"): self.records = records self.next_cursor = next_cursor + self.fetches = 0 async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: + self.fetches += 1 return FetchResult(records=list(self.records), next_cursor=self.next_cursor) +def _inbox_gate_context(db) -> GateContext: + return GateContext(job_id="inbox-processor", db=db) + + +async def _poll_inbox(db) -> str: + """What a ``poll_all_sources`` call of the inbox consumer reads.""" + result = await poll_all_sources_handler(ToolContext(session_id="s", db=db), {"consumer": "inbox"}) + return result.text_content + + @pytest.mark.asyncio async def test_insert_source_messages_rolls_back_the_batch_on_insert_failure(db): records = [_record("a"), _record("b"), _record("c")] @@ -202,7 +214,7 @@ async def test_a_new_source_s_first_records_are_unread_for_the_existing_consumer # The inbox consumer already reads another source. await db.insert_source_messages([dataclasses.replace(_record("x"), source="other")], source="other") await db.set_consumer_cursor("inbox", "other", await db.get_source_max_rowid("other")) - ctx = GateContext(job_id="inbox-processor", db=db) + ctx = _inbox_gate_context(db) assert not await MessagesGate().is_satisfied(ctx) result = await SourceRunner(_FixedSource([_record("a")]), db).run() @@ -232,9 +244,7 @@ async def test_a_baseline_run_seeds_the_cursor_so_later_records_wake_the_consume await runner.run() assert await db.consumer_has_unread("inbox") - text = (await poll_all_sources_handler( - ToolContext(session_id="s", db=db), {"consumer": "inbox"}, - )).text_content + text = await _poll_inbox(db) assert "Record later" in text assert f"{SOURCE_NAME}: 1 new" in text @@ -259,13 +269,26 @@ async def test_seeding_leaves_an_existing_cursor_and_only_runs_without_a_source_ @pytest.mark.asyncio -async def test_a_failure_to_start_the_consumer_cursors_does_not_stop_the_run(db, monkeypatch): +async def test_a_failure_to_start_the_consumer_cursors_fails_the_run_before_the_fetch(db, monkeypatch): + await db.set_consumer_cursor("inbox", "other", 0) + source = _FixedSource([_record("a")]) + runner = SourceRunner(source, db) + async def fail(source, ttl_days): raise RuntimeError("cursors unavailable") monkeypatch.setattr(db, "seed_consumer_cursors", fail) + result = await runner.run() - result = await SourceRunner(_FixedSource([_record("a")]), db).run() + assert result.records_ingested == 0 + assert "cursors unavailable" in result.error + assert source.fetches == 0 + assert await db.get_sync_cursor(SOURCE_NAME) is None - assert (result.records_ingested, result.error) == (1, None) - assert await db.get_sync_cursor(SOURCE_NAME) == "new" + # The next run tries again and delivers the first records. + monkeypatch.undo() + runner.health.backoff_until = None + assert (await runner.run()).records_ingested == 1 + for sources in ([], [SOURCE_NAME]): + assert await MessagesGate(sources=sources).is_satisfied(_inbox_gate_context(db)) + assert "Record a" in await _poll_inbox(db) From fe91cd4ccbfc06f73f96f7e54212916e71da1913 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 08:33:10 +0000 Subject: [PATCH 3/4] Sources: start an expired consumer cursor again before a first run Seeding kept every cursor that a consumer had on the new source, also an expired one. get_consumer_cursor treats an expired cursor as absent and starts it again at the source's newest message, so the first poll after the first run skipped the records that the run stored. Keep a live cursor on the new source. Start an expired one again at the newest message before the run, with a fresh TTL. Select every consumer that has a cursor, so that the expired cursor of a consumer that reads only the new source starts again too. Co-Authored-By: Claude Opus 5.5 --- docs/sources.md | 2 +- nerve/db/sources.py | 23 +++++++++++++------- tests/test_source_runner_persistence.py | 28 +++++++++++++++++++------ 3 files changed, 39 insertions(+), 14 deletions(-) diff --git a/docs/sources.md b/docs/sources.md index 9b09b94ff..c9c506d4e 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -55,7 +55,7 @@ Consumers read from the inbox using the `poll_source` agent tool with a named co **Key properties:** - Consumer cursors are independent — multiple consumers can read the same source at different positions - New consumer cursors initialize to the latest message (no backlog flooding) -- Before a source's first run, each consumer that has a cursor on another source gets a cursor on the new source, so the first records of the new source are unread for that consumer and its `messages` gates see them +- Before a source's first run, each consumer that has a cursor gets a live cursor on the new source at its newest message (a live cursor that is already there stays as it is), so the first records of the new source are unread for that consumer and its `messages` gates see them - Consumer cursors expire after N days of inactivity (configurable, default: 2 days); a `messages` gate that checks the consumer renews the cursors it reads (see [cron.md](cron.md#messages)) - Cross-source deduplication happens naturally because one agent session sees messages from all sources diff --git a/nerve/db/sources.py b/nerve/db/sources.py index ccfdfcb66..bda27a5e8 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -400,25 +400,34 @@ async def set_consumer_cursor( ) async def seed_consumer_cursors(self, source: str, ttl_days: int = 2) -> int: - """Start a cursor on *source* for every consumer that has one elsewhere. + """Start a cursor on *source* for every consumer that has a cursor. Each starts at the source's newest message, as :meth:`get_consumer_cursor` would, so what the source stores after this call is unread. Without it a new source stays invisible to :meth:`consumer_has_unread`, and the - consumer's first poll starts past the source's first records. A - consumer's existing cursor on *source* is left as it is. Returns the - number of cursors started. + consumer's first poll starts past the source's first records. A live + cursor on *source* is left as it is. An expired one starts again at + the newest message with a fresh TTL, because :meth:`get_consumer_cursor` + would start it again later, past the records stored in between. + Returns the number of cursors started. """ now = datetime.now(UTC) expires = (now + timedelta(days=ttl_days)).isoformat() + # SQLite needs a WHERE clause in the SELECT of an upsert. result = await self._write( - """INSERT OR IGNORE INTO consumer_cursors + """INSERT INTO consumer_cursors (consumer, source, cursor_seq, updated_at, expires_at) SELECT DISTINCT consumer, ?, (SELECT COALESCE(MAX(rowid), 0) FROM source_messages WHERE source = ?), ?, ? - FROM consumer_cursors WHERE source != ?""", - (source, source, now.isoformat(), expires, source), + FROM consumer_cursors WHERE true + ON CONFLICT(consumer, source) DO UPDATE SET + cursor_seq = excluded.cursor_seq, + updated_at = excluded.updated_at, + expires_at = excluded.expires_at + WHERE consumer_cursors.expires_at IS NOT NULL + AND consumer_cursors.expires_at <= excluded.updated_at""", + (source, source, now.isoformat(), expires), ) return result.rowcount or 0 diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index e27e416d3..9c506df3c 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -250,17 +250,20 @@ async def test_a_baseline_run_seeds_the_cursor_so_later_records_wake_the_consume @pytest.mark.asyncio -async def test_seeding_leaves_an_existing_cursor_and_only_runs_without_a_source_cursor(db): +async def test_seeding_leaves_a_live_cursor_and_only_runs_without_a_source_cursor(db): await db.insert_source_messages([_record("old")], source=SOURCE_NAME) + newest = await db.get_source_max_rowid(SOURCE_NAME) await db.set_consumer_cursor("inbox", SOURCE_NAME, 0) await db.set_consumer_cursor("other-consumer", "other", 0) + # A consumer whose only cursor, on the source, has expired. + await db.set_consumer_cursor("lapsed", SOURCE_NAME, 0, ttl_days=-1) - assert await db.seed_consumer_cursors(SOURCE_NAME) == 1 + assert await db.seed_consumer_cursors(SOURCE_NAME) == 2 assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 - # A consumer new to the source starts past what is already stored. - assert await db.peek_consumer_cursor("other-consumer", SOURCE_NAME) == ( - await db.get_source_max_rowid(SOURCE_NAME) - ) + # A consumer new to the source, or with an expired cursor on it, starts + # past what is already stored. + assert await db.peek_consumer_cursor("other-consumer", SOURCE_NAME) == newest + assert await db.peek_consumer_cursor("lapsed", SOURCE_NAME) == newest await db.set_sync_cursor(SOURCE_NAME, "stored") await db.set_consumer_cursor("late", "other", 0) @@ -268,6 +271,19 @@ async def test_seeding_leaves_an_existing_cursor_and_only_runs_without_a_source_ assert await db.peek_consumer_cursor("late", SOURCE_NAME) is None +@pytest.mark.asyncio +@pytest.mark.parametrize("sources", [[], [SOURCE_NAME]], ids=["any-source-gate", "new-source-gate"]) +async def test_an_expired_cursor_on_a_new_source_starts_again_before_its_first_records(db, sources): + await db.set_consumer_cursor("inbox", "other", 0) + await db.set_consumer_cursor("inbox", SOURCE_NAME, 0, ttl_days=-1) + + result = await SourceRunner(_FixedSource([_record("a")]), db).run() + + assert result.records_ingested == 1 + assert await MessagesGate(sources=sources).is_satisfied(_inbox_gate_context(db)) + assert "Record a" in await _poll_inbox(db) + + @pytest.mark.asyncio async def test_a_failure_to_start_the_consumer_cursors_fails_the_run_before_the_fetch(db, monkeypatch): await db.set_consumer_cursor("inbox", "other", 0) From 11823977b80ac6397c22496ded492bf59f3920c8 Mon Sep 17 00:00:00 2001 From: Alex Soffronow Pagonidis <237136924+alex-clickhouse@users.noreply.github.com> Date: Tue, 10 Nov 2026 11:54:57 +0000 Subject: [PATCH 4/4] Sources: start cursors only for consumers that read every source A consumer that reads only some sources with poll_source does not move a cursor on a new source. With a messages gate without sources, that cursor made the gate find unread messages on every check, and the gate's TTL renewal kept the cursor live. seed_consumer_cursors now starts a cursor only for a consumer that has a cursor on each other source with messages, as poll_all_sources leaves it. docs/cron.md tells a job that reads only some sources to name them in sources. Co-Authored-By: Claude Opus 5.5 --- docs/cron.md | 7 +++++++ docs/sources.md | 2 +- nerve/db/sources.py | 24 +++++++++++++++++++----- nerve/sources/runner.py | 3 ++- tests/test_source_runner_persistence.py | 24 ++++++++++++++++++++++++ 5 files changed, 53 insertions(+), 7 deletions(-) diff --git a/docs/cron.md b/docs/cron.md index 6566c20c7..3eceee97d 100644 --- a/docs/cron.md +++ b/docs/cron.md @@ -253,6 +253,13 @@ run_if: consumer: inbox ``` +Without `sources`, the gate checks the sources on which the consumer has a +cursor. A consumer that reads every source with `poll_all_sources` gets a cursor +on a new source before the source's first run, so the gate sees the first records +of the new source (see [sources.md](sources.md)). A consumer that reads only some +sources with `poll_source` does not get one. For such a job, name its sources in +`sources`. + Checking the gate keeps the consumer's cursors from expiring: each check renews the TTL (`sync.consumer_cursor_ttl_days`) of the live cursors it reads. A cursor expires when nothing reads it for that long, and an expired cursor starts again at diff --git a/docs/sources.md b/docs/sources.md index c9c506d4e..c0f2bf5a2 100644 --- a/docs/sources.md +++ b/docs/sources.md @@ -55,7 +55,7 @@ Consumers read from the inbox using the `poll_source` agent tool with a named co **Key properties:** - Consumer cursors are independent — multiple consumers can read the same source at different positions - New consumer cursors initialize to the latest message (no backlog flooding) -- Before a source's first run, each consumer that has a cursor gets a live cursor on the new source at its newest message (a live cursor that is already there stays as it is), so the first records of the new source are unread for that consumer and its `messages` gates see them +- Before a source's first run, each consumer that reads every source gets a live cursor on the new source at its newest message (a live cursor that is already there stays as it is), so the first records of the new source are unread for that consumer and its `messages` gates see them. Such a consumer has a cursor on each other source that has messages, as `poll_all_sources` leaves it. A consumer that reads only some sources with `poll_source` gets no cursor on the new source - Consumer cursors expire after N days of inactivity (configurable, default: 2 days); a `messages` gate that checks the consumer renews the cursors it reads (see [cron.md](cron.md#messages)) - Cross-source deduplication happens naturally because one agent session sees messages from all sources diff --git a/nerve/db/sources.py b/nerve/db/sources.py index bda27a5e8..595af69a7 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -400,7 +400,14 @@ async def set_consumer_cursor( ) async def seed_consumer_cursors(self, source: str, ttl_days: int = 2) -> int: - """Start a cursor on *source* for every consumer that has a cursor. + """Start a cursor on *source* for each consumer that reads every source. + + Such a consumer has a cursor on each other source that has messages, + as ``poll_all_sources`` leaves it. An expired cursor that is not yet + deleted also counts. A consumer that reads only some sources with + ``poll_source`` gets no cursor: its job would not move that cursor, so + a ``messages`` gate without ``sources`` would find unread messages on + every check. Each starts at the source's newest message, as :meth:`get_consumer_cursor` would, so what the source stores after this call is unread. Without it @@ -413,21 +420,28 @@ async def seed_consumer_cursors(self, source: str, ttl_days: int = 2) -> int: """ now = datetime.now(UTC) expires = (now + timedelta(days=ttl_days)).isoformat() - # SQLite needs a WHERE clause in the SELECT of an upsert. + # The WHERE clause also keeps SQLite from reading ON CONFLICT as a + # join constraint of the SELECT. result = await self._write( """INSERT INTO consumer_cursors (consumer, source, cursor_seq, updated_at, expires_at) - SELECT DISTINCT consumer, ?, + SELECT c.consumer, ?, (SELECT COALESCE(MAX(rowid), 0) FROM source_messages WHERE source = ?), ?, ? - FROM consumer_cursors WHERE true + FROM (SELECT DISTINCT consumer FROM consumer_cursors) AS c + WHERE NOT EXISTS ( + SELECT 1 + FROM (SELECT DISTINCT source FROM source_messages WHERE source != ?) AS s + WHERE NOT EXISTS ( + SELECT 1 FROM consumer_cursors AS cc + WHERE cc.consumer = c.consumer AND cc.source = s.source)) ON CONFLICT(consumer, source) DO UPDATE SET cursor_seq = excluded.cursor_seq, updated_at = excluded.updated_at, expires_at = excluded.expires_at WHERE consumer_cursors.expires_at IS NOT NULL AND consumer_cursors.expires_at <= excluded.updated_at""", - (source, source, now.isoformat(), expires), + (source, source, now.isoformat(), expires, source), ) return result.rowcount or 0 diff --git a/nerve/sources/runner.py b/nerve/sources/runner.py index e2b151433..2f9e69206 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -366,7 +366,8 @@ async def _run_locked(self) -> IngestResult: ) async def _seed_consumer_cursors(self) -> str | None: - """Before a source's first run, start the existing consumers' cursors on it. + """Before a source's first run, start a cursor on it for each consumer + that reads every source (see :meth:`SourceStore.seed_consumer_cursors`). This occurs before the source stores a record, so its first records are unread for those consumers, and their ``messages`` gates see them. diff --git a/tests/test_source_runner_persistence.py b/tests/test_source_runner_persistence.py index 9c506df3c..a0029d157 100644 --- a/tests/test_source_runner_persistence.py +++ b/tests/test_source_runner_persistence.py @@ -271,6 +271,30 @@ async def test_seeding_leaves_a_live_cursor_and_only_runs_without_a_source_curso assert await db.peek_consumer_cursor("late", SOURCE_NAME) is None +@pytest.mark.asyncio +async def test_seeding_skips_a_consumer_that_reads_only_some_sources(db): + for name in ("alpha", "beta"): + await db.insert_source_messages([dataclasses.replace(_record("x"), source=name)], source=name) + alpha, beta = await db.get_source_max_rowid("alpha"), await db.get_source_max_rowid("beta") + # "inbox" reads every source. "alpha-only" polls only alpha. "lapsed" reads + # every source, but its cursor on beta has expired. + await db.set_consumer_cursor("inbox", "alpha", alpha) + await db.set_consumer_cursor("inbox", "beta", beta) + await db.set_consumer_cursor("alpha-only", "alpha", alpha) + await db.set_consumer_cursor("lapsed", "alpha", alpha) + await db.set_consumer_cursor("lapsed", "beta", beta, ttl_days=-1) + + await SourceRunner(_FixedSource([_record("a")]), db).run() + + assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 + assert await db.peek_consumer_cursor("lapsed", SOURCE_NAME) == 0 + assert await db.peek_consumer_cursor("alpha-only", SOURCE_NAME) is None + # The gate of a job that polls only alpha does not see the new source. + ctx = _inbox_gate_context(db) + assert not await MessagesGate(consumer="alpha-only").is_satisfied(ctx) + assert await MessagesGate().is_satisfied(ctx) + + @pytest.mark.asyncio @pytest.mark.parametrize("sources", [[], [SOURCE_NAME]], ids=["any-source-gate", "new-source-gate"]) async def test_an_expired_cursor_on_a_new_source_starts_again_before_its_first_records(db, sources):