From b0c9b45d985a14b9807737ddb546b22e41c0a079 Mon Sep 17 00:00:00 2001 From: Minh Vu Date: Sat, 1 Aug 2026 17:45:16 +0200 Subject: [PATCH] github_repos: preserve cursor after partial fetch failures --- nerve/sources/github_repos.py | 25 ++++++++--- tests/test_github_repos_source.py | 69 +++++++++++++++++++++++++++++++ 2 files changed, 89 insertions(+), 5 deletions(-) diff --git a/nerve/sources/github_repos.py b/nerve/sources/github_repos.py index 76834b42..3663e853 100644 --- a/nerve/sources/github_repos.py +++ b/nerve/sources/github_repos.py @@ -71,9 +71,15 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: # Flatten into (repo, issue) pairs, skipping repos that errored. items: list[tuple[str, dict]] = [] + fetch_failed = False for repo, res in zip(self._repos, results): if isinstance(res, Exception): logger.warning("github_repos: fetch failed for %s: %s", repo, res) + fetch_failed = True + continue + if res is None: + logger.warning("github_repos: fetch failed for %s", repo) + fetch_failed = True continue for issue in res: if isinstance(issue, dict): @@ -84,6 +90,11 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: # First run: establish baseline cursor, don't backfill history. if cursor is None: + if fetch_failed: + logger.info( + "github_repos: first run incomplete, baseline not established", + ) + return FetchResult(records=[], next_cursor=None) newest_ts = max( (it.get("created_at", "") for _, it in items), default=None, @@ -108,9 +119,13 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: new_items.sort(key=lambda ri: ri[1].get("created_at", "")) records = [self._issue_to_record(repo, it) for repo, it in new_items] - # Advance cursor to the newest created_at we just ingested. + # Successful repositories can be ingested immediately, but a shared + # cursor is safe to advance only after every repository was observed. + # Unchanged retries from healthy repositories are deduplicated by + # (source, id). newest_ts = max(it.get("created_at", "") for _, it in new_items) - return FetchResult(records=records, next_cursor=newest_ts, has_more=False) + next_cursor = cursor if fetch_failed else newest_ts + return FetchResult(records=records, next_cursor=next_cursor, has_more=False) # ------------------------------------------------------------------ # Fetch + formatting helpers @@ -118,11 +133,11 @@ async def fetch(self, cursor: str | None, limit: int = 100) -> FetchResult: async def _fetch_repo( self, repo: str, per_page: int, sem: asyncio.Semaphore, - ) -> list[dict]: + ) -> list[dict] | None: """Fetch the most-recently-created issues+PRs for a single repo. Returns a list of issue dicts (PRs included — they carry a - ``pull_request`` key). Returns [] on any error. + ``pull_request`` key), or ``None`` on error. """ endpoint = ( f"repos/{repo}/issues" @@ -131,7 +146,7 @@ async def _fetch_repo( async with sem: data = await self._gh_api_get(endpoint) if not isinstance(data, list): - return [] + return None return data def _issue_to_record(self, repo: str, issue: dict) -> SourceRecord: diff --git a/tests/test_github_repos_source.py b/tests/test_github_repos_source.py index 375e6574..1747709e 100644 --- a/tests/test_github_repos_source.py +++ b/tests/test_github_repos_source.py @@ -165,6 +165,75 @@ async def flaky_gh_api_get(endpoint, timeout=30): repos_seen = {r.metadata["repo_name"] for r in result.records} assert repos_seen == {"owner/repo-a"} assert len(result.records) == 2 + assert result.next_cursor == "2026-06-12T08:00:00Z" + + +@pytest.mark.asyncio +async def test_partial_failure_keeps_cursor_until_missed_item_can_be_fetched(): + src = GitHubReposSource(config={"repos": ["owner/repo-a", "owner/repo-b"]}) + repo_b_available = False + data = { + "owner/repo-a": [ + _issue(1, 101, "2026-06-12T11:00:00Z", "Issue A"), + ], + "owner/repo-b": [ + _issue(2, 102, "2026-06-12T10:30:00Z", "Issue B"), + ], + } + + async def flaky_gh_api_get(endpoint, timeout=30): + match = re.match(r"repos/([^/]+/[^/]+)/issues", endpoint) + repo = match.group(1) + if repo == "owner/repo-b" and not repo_b_available: + return None + return data[repo] + + src._gh_api_get = flaky_gh_api_get + + first = await src.fetch(cursor="2026-06-12T10:00:00Z") + + assert [record.id for record in first.records] == ["101"] + assert first.next_cursor == "2026-06-12T10:00:00Z" + + repo_b_available = True + second = await src.fetch(cursor=first.next_cursor) + + assert [record.id for record in second.records] == ["102", "101"] + assert second.next_cursor == "2026-06-12T11:00:00Z" + + +@pytest.mark.asyncio +async def test_first_run_waits_for_every_repository_before_setting_baseline(): + src = GitHubReposSource(config={"repos": ["owner/repo-a", "owner/repo-b"]}) + + async def flaky_gh_api_get(endpoint, timeout=30): + if "repo-b" in endpoint: + return None + return _REPO_DATA["owner/repo-a"] + + src._gh_api_get = flaky_gh_api_get + result = await src.fetch(cursor=None) + + assert result.records == [] + assert result.next_cursor is None + + +@pytest.mark.asyncio +async def test_empty_repository_response_does_not_prevent_cursor_advance(): + src = _make_source( + ["owner/repo-a", "owner/repo-b"], + data={ + "owner/repo-a": [ + _issue(1, 101, "2026-06-12T11:00:00Z", "Issue A"), + ], + "owner/repo-b": [], + }, + ) + + result = await src.fetch(cursor="2026-06-12T10:00:00Z") + + assert [record.id for record in result.records] == ["101"] + assert result.next_cursor == "2026-06-12T11:00:00Z" @pytest.mark.asyncio