Repository navigation
Sources: start consumer cursors before a source's first run - #677
alex-clickhouse wants to merge 4 commits into
Conversation
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
|
Nerve PR review · ✅ Ready to merge Updated for 1182397: the seed now starts a cursor only for a consumer that has a cursor on every other source with messages, which is the state The PR fixes a real gap: the first records of a new source ( Completeness: tests cover the first records, a baseline run, live and expired cursors, a failed seed, the hosted drain and the narrowing. Worth considering:
Checked: 7 source/gate/DB test files in an offline sandbox at |
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 <noreply@anthropic.com>
|
Thanks. Fixed in 1182397, with the first option and a docs note:
One limit stays: when the consumer's sources are the only sources with messages, a consumer that polls them one by one looks the same as one that uses |
What
Before a source's first run, give each consumer that reads every source a
live cursor on the source, at the source's newest message. The first
records of the new source are then unread for those consumers, and their
messagesgates see them.Why
A consumer that reads other sources does not see the first records of a
new source:
consumer_has_unreadignores a source for which the consumer has nocursor. A
messagesgate withoutsourcesuses it, so the gate does notfire for the new source.
message, past the first records. An expired cursor on the source starts
again in the same way.
This affects
github_reposand every source that starts after theconsumers have cursors, also the hosted sources.
How
SourceStore.seed_consumer_cursors(source)upserts a cursor on thesource for each consumer that reads every source. Such a consumer has a
cursor on each other source that has messages, as
poll_all_sourcesleaves it. An expired cursor that is not yet deleted also counts.
poll_sourcegets nocursor. Its job does not move a cursor on the new source, so a
messagesgate without
sourceswould find unread messages on every check, and thegate's TTL renewal would keep that cursor live.
the source starts again at the newest message with a fresh TTL. A live
cursor stays as it is.
SourceRunnercalls it before the first fetch when the source has nosync cursor. The TTL is
sync.consumer_cursor_ttl_days.failure. The source stores nothing and has no sync cursor, so the next
run tries again.
docs/sources.mddescribes the behavior.docs/cron.mdtells a jobthat reads only some sources to name them in
sources.Based on a fix by Frank Rosner.
Tests
messagesgate on any source and on the new source sees them.reach
poll_all_sources.at the newest message, and the seeding does not run for a source with a
sync cursor.
source, and its gate without
sourcesstays false. A consumer with anexpired cursor on one of the other sources gets a cursor.
gate and
poll_all_sources.next run delivers the first records.
github_reposbuffer wakes themessagesgate of a consumer that reads
github.test_platform_activity.py::test_a_burst_of_changes_is_coalesced, a timing test, at load average 11; it passes alone, 3 runs out of 3).🤖 Generated with Claude Code