GitHub Repos source: report the last day after a gap, not nothing

config.example.yaml

@@ -343,9 +343,10 @@ sync:
   # Repositories you follow without taking part in them: new PRs, issues and
   # releases there. Nothing is set on the GitHub account. With
   # sync.github.mode: hosted the channel gateway polls them too (it must know
-  # the repository watch). Nothing older than a day is backfilled: a long
-  # outage, or the watch switched off for more than a day (in hosted mode,
-  # switched off at all), starts fresh. See docs/sources.md "GitHub Repos".
+  # the repository watch). Nothing older than a day is backfilled: a check
+  # more than a day after the last (a daily schedule, an outage) reports the
+  # last day's items, with an hour of grace; in hosted mode, switching the
+  # watch off and on starts fresh. See docs/sources.md "GitHub Repos".
   github_repos:
     enabled: false
     schedule: "*/15 * * * *"       # poll mode; hosted: the fallback drain

docs/config.md

@@ -1831,7 +1831,7 @@ reload:
 |-----|------|---------|-------------|
 | `sync.github_repos.enabled` | bool | `false` | Enable the repository watch. In hosted mode Nerve publishes it to the gateway only while it is on and `repos` lists a repository |
 | `sync.github_repos.schedule` | cron | `*/15 * * * *` | Poll frequency; in hosted mode the fallback cadence of the drain, which also runs on every delivery |
-| `sync.github_repos.repos` | list | `[]` | `owner/repo` names as GitHub writes them, each once (case does not matter), at most 50. While the watch is on, a name GitHub would not take, a repository named twice, or more than 50 refuses to load; while it is off, a warning names them. Empty = nothing is watched. Nothing older than a day is backfilled: a long outage, or the watch switched off for more than a day (in hosted mode, switched off at all), starts fresh |
+| `sync.github_repos.repos` | list | `[]` | `owner/repo` names as GitHub writes them, each once (case does not matter), at most 50. While the watch is on, a name GitHub would not take, a repository named twice, or more than 50 refuses to load; while it is off, a warning names them. Empty = nothing is watched. Nothing older than a day is backfilled: a check more than a day after the last one (a daily schedule, an outage, the watch switched off for longer) reports the last day's items, with an hour of grace; in hosted mode, switching the watch off and on starts fresh |
 | `sync.github_repos.types` | list | `[pull_request, issue]` | What is reported: `pull_request`, `issue`, `release`. Case does not matter; an unknown type or an empty list refuses to load |
 | `sync.github_repos.include_bots` | bool | `false` | Also report the pull requests and issues that a GitHub App or an account whose login ends in `[bot]` opened. A release is reported whoever published it (release automation such as `github-actions[bot]` is the normal case) |
 | `sync.github_repos.include_drafts` | bool | `true` | `false`: a draft pull request is reported when it is marked ready for review (event `ready`), not when it is opened |

docs/sources.md

@@ -315,9 +315,9 @@ notifications, and in hosted mode through `pr_watch`.
 - **One page each:** a repository with more than 100 pull requests (or issues) changed between two runs loses the oldest of those changes. `batch_size` does not cap this source
 - **Cursor:** JSON. Per repository and listing: the newest `updated_at` listed (`published_at` for releases), when the listing was read, and the numbers (release ids) of the items created in that newest second. Per repository, the pull requests it follows: a held draft (`include_drafts: false`) and, with `updates`, a reported pull request while it is open, as head commit, draft and held. At most 1,000 per repository; the one listed longest ago is forgotten first
 - **First run:** the baseline. Each listing's cursor is set and nothing is reported, so switching the source on does not fill the inbox with history; a repository or a type added later takes its own baseline, and one removed is forgotten. An item is new when it was created (a release: published) after the last listing: later than the listing's cursor, or in the cursor's own second and not listed then, since GitHub's times have a resolution of one second. So an older pull request that changes later is never reported as new, and an item is reported once
-- **No backfill:** nothing older than a day is backfilled. A listing last read more than a day ago (the watch switched off for longer, or a long outage) starts fresh with a new baseline, and the pull requests it followed are forgotten
+- **No backfill beyond a day:** nothing older than a day is backfilled, and the last day is. A check more than a day after the last one (a daily schedule, which always comes a moment past the day; a weekday schedule after the weekend; the watch switched off for longer; a long outage) reads the listing as of a day and an hour ago: it reports the items of the last day, with an hour of grace for the scheduler's delay (an item created moments after the last check is not lost), within the one page, and nothing older, and the pull requests it follows stay followed. So switching the watch off and on in poll mode resumes where it left within the day, and beyond it reports the last day's items
 - **Failures:** a listing that fails (a repository that does not exist or that the `gh` login cannot read, a rate limit) keeps its cursor and is read again on the next run; the other listings still report. A run in which every listing fails fails, and the source's health shows it
-- **Upgrades:** the cursor of the earlier version of this source, one timestamp, becomes the cursor of every listing (unless it is more than a day old), so nothing is repeated or skipped; a buffer position left by the hosted mode is no cursor (a new baseline). The earlier version took any `repos` value: one that the watch now refuses (a URL such as `https://github.com/acme/widgets`, a trailing slash, `owner/repo/x`, the same repository twice, more than 50) only logs a warning while the watch is off, and refuses the config once it is on
+- **Upgrades:** the cursor of the earlier version of this source, one timestamp, becomes the cursor of every listing (one more than a day old reports the last day), so nothing is repeated or skipped; a buffer position left by the hosted mode is no cursor (a new baseline). The earlier version took any `repos` value: one that the watch now refuses (a URL such as `https://github.com/acme/widgets`, a trailing slash, `owner/repo/x`, the same repository twice, more than 50) only logs a warning while the watch is off, and refuses the config once it is on
 - **Default schedule:** `*/15 * * * *` (every 15 min); in hosted mode the fallback cadence of the drain
 - **Disabled by default**
 
@@ -411,10 +411,10 @@ machine for them.
   `deny_authors` list the gateway cannot take (more than 256 patterns, or a
   pattern longer than 256 bytes or with a control character) goes with the
   patterns that fit, and the inbox guardrail here applies them all.
-- **It starts fresh.** The gateway backfills nothing older than a day either,
-  and it forgets the repositories of an agent that stops subscribing to them,
-  so switching the watch off and on, or a long outage, starts from a new
-  baseline.
+- **It starts fresh when switched off.** The gateway backfills nothing older
+  than a day either (after a gap it reports the last day), and it forgets the
+  repositories of an agent that stops subscribing to them, so switching the
+  watch off and on in hosted mode starts from a new baseline.
 - **The gateway must know the repository watch.** The subscription carries a
   `repos` member that a gateway older than the repository watch refuses: it
   closes the stream, hosted Slack traffic included. Update the gateway before

nerve/sources/github_repos.py

@@ -53,12 +53,16 @@
 listed longest ago is forgotten first. A listing without a cursor is the
 baseline: its cursor is set and nothing is reported, so switching the source
 on, or adding a repository or a type, does not fill the inbox with history.
-So is a listing last read more than a day ago (:data:`STALE_AFTER`): nothing
-older than a day is backfilled, and a long outage, or the source switched off
-for longer, starts fresh. A listing that fails keeps its cursor. The earlier
-cursor of this source, one ISO timestamp (the newest ``created_at`` seen),
-stands for every listing and for when it was last read; a buffer position
-(digits), left by the hosted mode, is no cursor.
+Nothing older than a day is backfilled: a listing last read more than a day
+ago (:data:`STALE_AFTER`; a daily schedule, a long outage, the source switched
+off for longer) is read as of a day and an hour ago (:data:`STALE_GRACE`, for
+the scheduler's delay), so it reports what is new in the last day, with an
+hour of grace, and nothing older, and the pull requests it follows stay
+followed. A
+listing that fails keeps its cursor. The earlier cursor of this source, one
+ISO timestamp (the newest ``created_at`` seen), stands for every listing and
+for when it was last read; a buffer position (digits), left by the hosted
+mode, is no cursor.
 """
 
 from __future__ import annotations
@@ -100,10 +104,13 @@
 
 _CURSOR_VERSION = 1
 
-# A listing last read longer ago than this starts again from a baseline:
-# nothing older than a day is backfilled, the horizon of the notifications
-# source's first run too.
+# A listing last read longer ago than this is read as of this long ago, and
+# STALE_GRACE more: nothing older than a day is backfilled (the horizon of the
+# notifications source's first run too), and the last day is, with an hour of
+# grace for the scheduler's delay. A daily check comes back a day and a moment
+# after the last one, and an item created in that moment is not lost.
 STALE_AFTER = timedelta(hours=24)
+STALE_GRACE = timedelta(hours=1)
 
 # What each type lists: its name in the cursor, and the endpoint.
 _LISTINGS: dict[str, tuple[str, str]] = {
@@ -317,6 +324,27 @@ def _stale(listing: _Listing, now: datetime) -> bool:
     return when is None or now - when > STALE_AFTER
 
 
+def _window(listing: _Listing, now: datetime) -> _Listing:
+    """The cursor to read a stale *listing* against: the last day, with an hour of grace.
+
+    A daily schedule comes back a day and a moment later, a weekday one after
+    the weekend: what is new in the last day (and the hour of
+    :data:`STALE_GRACE` before it, for the scheduler's delay) is reported,
+    nothing older. The cursor is *listing*'s own when it falls within that
+    window (its keys still name what it saw in its second); otherwise the
+    window's start, with no keys. Never earlier than *listing*'s own cursor,
+    so nothing is reported twice: nothing listed back then was created after
+    that cursor.
+    """
+    # Cut to the second first, as the stored stamp is, so a cursor on exactly
+    # that second keeps its keys (the gateway compares the same way).
+    start = _stamp(now - STALE_AFTER - STALE_GRACE)
+    since = _when(listing.since)
+    if since is not None and since >= _when(start):
+        return listing
+    return _Listing(since=start, listed_at=listing.listed_at, at=[])
+
+
 def _number_key(item: dict[str, Any]) -> str:
     return str(item.get("number"))
 
@@ -656,15 +684,15 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult:
                         state.listings[name] = previous
                     continue
                 if previous is not None and _stale(previous, now):
-                    # Nothing older than a day is backfilled: after a long
-                    # outage, or a watch switched off for longer, start fresh.
+                    # Nothing older than a day is backfilled, and the last day
+                    # is: a daily schedule is always a moment past the day.
+                    # The pull requests it follows stay followed.
                     logger.info(
                         "github_repos: %s %s was last read at %s, more than a day ago; "
-                        "taking a new baseline", repo, name, previous.listed_at or "an unknown time",
+                        "reporting what is new in the last day",
+                        repo, name, previous.listed_at or "an unknown time",
                     )
-                    previous = None
-                    if kind == "pull_request":
-                        state.prs.clear()
+                    previous = _window(previous, now)
                 if kind == "pull_request":
                     records += self._pull_requests(repo, result, previous, state, now)
                 elif kind == "issue":

tests/test_github_repos_source.py

@@ -333,29 +333,123 @@ async def test_a_release_published_in_the_cursors_second_after_the_listing_is_re
     assert (await src.fetch(result.next_cursor)).records == []
 
 
+# Minutes in a day, for times given as minutes after START (NOW is at(60)).
+DAY = 24 * 60
+
+
 @pytest.mark.asyncio
-async def test_a_listing_read_more_than_a_day_ago_starts_fresh():
-    """Nothing older than a day is backfilled: after a long outage, or with
-    the watch switched off for longer, the next listing is a new baseline,
-    and the pull requests it followed are forgotten."""
+async def test_a_check_a_day_and_a_second_after_the_last_reports_the_last_day():
+    """A daily schedule comes back a day and a moment after the last check.
+    Nothing older than a day is backfilled, but the last day is: what was
+    opened since the last check is reported; an older item that only
+    changed is not."""
     fake = FakeGitHub()
+    fake.issues[REPO] = [issue(1, created=-600)]
     now = [NOW]
-    src = source(fake, types=["pull_request", "issue"], updates=True, clock=lambda: now[0])
+    src = source(fake, types=["pull_request", "issue"], clock=lambda: now[0])
     cursor = await baseline(src)
-    fake.pulls[REPO] = [pull(1, created=5)]
-    cursor = (await src.fetch(cursor)).next_cursor
-    assert json.loads(cursor)["repos"][REPO]["prs"] == {"1": [SHA_A[:12], False, False]}
-    fake.pulls[REPO] = [pull(1, created=5, updated=1490, sha=SHA_B), pull(2, created=1495)]
-    fake.issues[REPO] = [issue(3, created=1496)]
-    now[0] = NOW + timedelta(hours=24, minutes=1)
+    fake.pulls[REPO] = [pull(2, created=120)]
+    fake.issues[REPO] = [issue(1, created=-600, updated=1000), issue(3, created=1260)]
+    now[0] = NOW + timedelta(hours=24, seconds=1)
 
     result = await src.fetch(cursor)
 
-    assert result.records == []
-    stamp = now[0].strftime("%Y-%m-%dT%H:%M:%SZ")
-    assert cursor_of(result) == {"v": 1, "repos": {REPO: {
-        "pulls": read(at(1495), "2", listed_at=stamp), "issues": read(at(1496), "3", listed_at=stamp),
-    }}}
+    assert [record.id for record in result.records] == [f"{REPO}#2:opened:open", f"{REPO}#3:opened:open"]
+    assert (await src.fetch(result.next_cursor)).records == []
+
+
+@pytest.mark.asyncio
+async def test_a_check_a_day_and_five_seconds_after_the_last_keeps_an_item_of_the_moment_after_it():
+    """The window reaches an hour further back than the day, for the
+    scheduler's delay: a pull request created 2 s after the last check, which
+    that check did not list, is still reported; the one it listed is not."""
+    fake = FakeGitHub()
+    fake.pulls[REPO] = [pull(1, created=-600)]
+    now = [NOW]
+    src = source(fake, types=["pull_request"], clock=lambda: now[0])
+    cursor = await baseline(src)
+    moment_after = pull(2, created=60)
+    moment_after["created_at"] = moment_after["updated_at"] = (
+        NOW + timedelta(seconds=2)
+    ).strftime("%Y-%m-%dT%H:%M:%SZ")
+    fake.pulls[REPO] = [pull(1, created=-600), moment_after]
+    now[0] = NOW + timedelta(hours=24, seconds=5)
+
+    result = await src.fetch(cursor)
+
+    assert [record.id for record in result.records] == [f"{REPO}#2:opened:open"]
+
+
+@pytest.mark.asyncio
+async def test_a_check_a_week_after_the_last_reports_only_the_last_day():
+    fake = FakeGitHub()
+    now = [NOW]
+    src = source(fake, types=["pull_request", "issue", "release"], clock=lambda: now[0])
+    cursor = await baseline(src)
+    fake.pulls[REPO] = [
+        pull(1, created=60 + DAY),
+        pull(2, created=60 + 6 * DAY + 720),
+        pull(4, created=60 + 6 * DAY - 30),
+        pull(5, created=60 + 6 * DAY - 120),
+    ]
+    fake.issues[REPO] = [issue(3, created=60 + 6 * DAY + 1380)]
+    fake.releases[REPO] = [
+        release(900, tag="v1", created=60 + 3 * DAY, published=60 + 3 * DAY),
+        release(901, tag="v2", created=60 + 6 * DAY + 1200, published=60 + 6 * DAY + 1200),
+    ]
+    now[0] = NOW + timedelta(days=7)
+
+    result = await src.fetch(cursor)
+
+    # Six days, four days and 26 h old: not reported. 24.5 h old (within the
+    # hour of grace), 12, 4 and 1 h old: reported.
+    assert [record.id for record in result.records] == [
+        f"{REPO}#4:opened:open", f"{REPO}#2:opened:open", f"{REPO}#:released:901", f"{REPO}#3:opened:open",
+    ]
+
+
+@pytest.mark.asyncio
+async def test_after_a_gap_the_followed_pull_requests_stay_followed():
+    """updates with a daily schedule: a followed pull request that changed in
+    the last day is reported after the gap, as it is without one."""
+    fake = FakeGitHub()
+    now = [NOW]
+    src = source(fake, types=["pull_request"], updates=True, clock=lambda: now[0])
+    cursor = await baseline(src)
+    fake.pulls[REPO] = [pull(1, created=90)]
+    now[0] = NOW + timedelta(hours=1)
+    opened = await src.fetch(cursor)
+    assert [record.id for record in opened.records] == [f"{REPO}#1:opened:open"]
+    fake.pulls[REPO] = [pull(1, created=90, updated=60 + 2 * DAY, sha=SHA_B)]
+    now[0] = NOW + timedelta(days=2, hours=1)
+
+    result = await src.fetch(opened.next_cursor)
+
+    assert [record.id for record in result.records] == [f"{REPO}#1:updated:{SHA_B}"]
+
+
+@pytest.mark.asyncio
+async def test_a_listing_whose_reading_time_is_unknown_is_read_against_the_last_day():
+    fake = FakeGitHub()
+    fake.pulls[REPO] = [pull(1, created=-3000), pull(2, created=5)]
+    cursor = json.dumps({"v": 1, "repos": {REPO: {"pulls": {"since": at(-4000), "at": []}}}})
+
+    result = await source(fake, types=["pull_request"]).fetch(cursor)
+
+    assert [record.id for record in result.records] == [f"{REPO}#2:opened:open"]
+
+
+@pytest.mark.asyncio
+async def test_a_stale_listing_whose_cursor_falls_within_the_day_keeps_its_keys():
+    """Its cursor second is still the window's: the item it saw there is not
+    reported again, the one it did not see is."""
+    fake = FakeGitHub()
+    fake.pulls[REPO] = [pull(1, created=5), pull(2, created=5)]
+    cursor = json.dumps({"v": 1, "repos": {REPO: {"pulls": {"since": at(5), "at": ["1"]}}}})
+
+    result = await source(fake, types=["pull_request"]).fetch(cursor)
+
+    assert [record.id for record in result.records] == [f"{REPO}#2:opened:open"]
 
 
 @pytest.mark.asyncio
@@ -830,20 +924,19 @@ async def test_the_cursor_of_the_earlier_source_still_reads():
 
 
 @pytest.mark.asyncio
-async def test_the_cursor_of_the_earlier_source_more_than_a_day_old_starts_fresh():
+async def test_the_cursor_of_the_earlier_source_more_than_a_day_old_reports_the_last_day():
+    """24.5 h old is within the day and its hour of grace; 25 h 55 min old is not."""
     fake = FakeGitHub()
-    fake.pulls[REPO] = [pull(1, created=5)]
-    src = source(fake, types=["pull_request"], clock=lambda: START + timedelta(hours=25))
+    fake.pulls[REPO] = [pull(1, created=5), pull(2, created=90)]
+    src = source(fake, types=["pull_request"], clock=lambda: START + timedelta(hours=26))
 
-    assert (await src.fetch(at(0))).records == []
+    result = await src.fetch(at(0))
+
+    assert [record.id for record in result.records] == [f"{REPO}#2:opened:open"]
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("cursor", [
-    "57", "not a cursor", "{broken json", '{"v": 1, "repos": []}',
-    # A listing whose reading time is unknown counts as read long ago.
-    json.dumps({"v": 1, "repos": {REPO: {"pulls": {"since": at(0), "at": []}}}}),
-])
+@pytest.mark.parametrize("cursor", ["57", "not a cursor", "{broken json", '{"v": 1, "repos": []}'])
 async def test_a_cursor_that_cannot_be_read_takes_a_baseline(cursor):
     """A buffer position left by the hosted mode, or anything unreadable,
     is no cursor: nothing is reported."""