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 6b50d6083..c0f2bf5a2 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 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 66cdaab95..595af69a7 100644 --- a/nerve/db/sources.py +++ b/nerve/db/sources.py @@ -399,6 +399,52 @@ 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 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 + a new source stays invisible to :meth:`consumer_has_unread`, and the + 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() + # 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 c.consumer, ?, + (SELECT COALESCE(MAX(rowid), 0) FROM source_messages WHERE source = ?), + ?, ? + 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), + ) + 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..2f9e69206 100644 --- a/nerve/sources/runner.py +++ b/nerve/sources/runner.py @@ -322,6 +322,10 @@ async def _run_locked(self) -> IngestResult: there. """ cursor = await self.db.get_sync_cursor(self.source.source_name) + if cursor is None: + error = await self._seed_consumer_cursors() + if error: + return IngestResult(records_ingested=0, error=error) total_ingested = 0 total_dropped = 0 @@ -361,6 +365,35 @@ async def _run_locked(self) -> IngestResult: records_ingested=total_ingested, records_dropped=total_dropped, ) + async def _seed_consumer_cursors(self) -> str | None: + """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. + 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 + + try: + seeded = await self.db.seed_consumer_cursors( + self.source.source_name, ttl_days=_consumer_cursor_ttl_days(), + ) + 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 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.""" 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..a0029d157 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 @@ -28,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")] @@ -192,3 +207,128 @@ 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 = _inbox_gate_context(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_inbox(db) + assert "Record later" in text + assert f"{SOURCE_NAME}: 1 new" in text + + +@pytest.mark.asyncio +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) == 2 + assert await db.peek_consumer_cursor("inbox", SOURCE_NAME) == 0 + # 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) + await SourceRunner(_FixedSource([]), db).run() + 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): + 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) + 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() + + 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 + + # 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)