From 5ad595b10ae1072f7547e1f9f1282151af55ea4b Mon Sep 17 00:00:00 2001 From: anderdc Date: Tue, 16 Jun 2026 17:43:34 -0500 Subject: [PATCH] feat(mirror-client): page miner pulls/issues fetches via next_cursor High-volume miners' windowed POST /miners/:id/pulls (and /issues) fetch runs the inlined per-row subqueries over the miner's whole PR set in one shot, exceeding MIRROR_HTTP_TIMEOUT_SECONDS; the miner is then skipped onto a stale cached evaluation and new PRs never ingest. Follow the mirror's cursor pagination (now on the windowed POST paths, das-github-mirror#191) so each page bounds the server-side query cost. _fetch_paginated sends limit on every request and cursor once a next_cursor comes back, concatenating each page's rows; a mirror that predates windowed pagination ignores the params and returns no next_cursor, so the loop completes in one request and behavior is unchanged. MIRROR_MAX_PAGES caps runaway paging and logs if hit. Depends on das-github-mirror#191 being deployed to mirror.gittensor.io first; until then this is a safe no-op (single unbounded request). --- gittensor/constants.py | 6 ++ gittensor/utils/mirror/client.py | 71 +++++++++++++--- tests/utils/test_mirror_client.py | 135 ++++++++++++++++++++++++++++++ 3 files changed, 201 insertions(+), 11 deletions(-) diff --git a/gittensor/constants.py b/gittensor/constants.py index e513dbec3..80a59d8c2 100644 --- a/gittensor/constants.py +++ b/gittensor/constants.py @@ -33,6 +33,12 @@ # File endpoint returns head/base blob contents; allow more time than plain GitHub calls. MIRROR_HTTP_TIMEOUT_SECONDS = 30 MIRROR_MAX_ATTEMPTS = 3 +# Miner pulls/issues list endpoints are cursor-paginated; page through next_cursor +# at this size. 100 keeps each page's server-side subquery cost well under +# MIRROR_HTTP_TIMEOUT_SECONDS even for the highest-volume miners. +MIRROR_PAGE_LIMIT = 100 +# Defensive cap on pages followed per miner list fetch (100 * 100 = 10k rows). +MIRROR_MAX_PAGES = 100 # ============================================================================= # Language & File Scoring diff --git a/gittensor/utils/mirror/client.py b/gittensor/utils/mirror/client.py index d365ef8cb..053565f9e 100644 --- a/gittensor/utils/mirror/client.py +++ b/gittensor/utils/mirror/client.py @@ -16,6 +16,8 @@ GITTENSOR_MIRROR_DEFAULT_URL, MIRROR_HTTP_TIMEOUT_SECONDS, MIRROR_MAX_ATTEMPTS, + MIRROR_MAX_PAGES, + MIRROR_PAGE_LIMIT, ) from gittensor.utils.mirror.models import ( MirrorIssuesResponse, @@ -74,7 +76,7 @@ def get_miner_pulls( window across all tracked repos. """ path = f'/api/v1/miners/{github_id}/pulls' - data = self._fetch_windowed(path, since_by_repo) + data = self._fetch_windowed(path, 'pull_requests', since_by_repo) try: return MirrorPullRequestsResponse.from_dict(data) except Exception as e: @@ -93,7 +95,7 @@ def get_miner_issues( open-issue-count path. """ path = f'/api/v1/miners/{github_id}/issues' - data = self._fetch_windowed(path, since_by_repo) + data = self._fetch_windowed(path, 'issues', since_by_repo) try: return MirrorIssuesResponse.from_dict(data) except Exception as e: @@ -130,22 +132,69 @@ def get_repo_maintainers(self, repo_full_name: str) -> MirrorRepoMaintainersResp except Exception as e: raise MirrorRequestError(f'Mirror response from {path} was invalid: {e}') from e - def _fetch_windowed(self, path: str, since_by_repo: Optional[Dict[str, datetime]]) -> dict: - """POST a per-repo ``since`` map when one is given, else GET the - mirror's default window.""" + def _fetch_windowed( + self, + path: str, + list_key: str, + since_by_repo: Optional[Dict[str, datetime]], + ) -> dict: + """Fetch a miner list endpoint, following pagination to completion. + + POSTs a per-repo ``since`` map when one is given, else GETs the mirror's + default window. Either way the response is paged via ``next_cursor`` and + the per-page ``list_key`` arrays are concatenated into one response dict. + """ if since_by_repo: body = { 'since_by_repo': {repo: dt.astimezone(timezone.utc).isoformat() for repo, dt in since_by_repo.items()} } - return self._post(path, body) - return self._get(path) + return self._fetch_paginated('POST', path, list_key, json_body=body) + return self._fetch_paginated('GET', path, list_key) + + def _fetch_paginated( + self, + method: str, + path: str, + list_key: str, + json_body: Optional[dict] = None, + ) -> dict: + """Page through ``path``, concatenating each page's ``list_key`` rows. + + Sends ``limit`` on every request and ``cursor`` once the mirror returns + a ``next_cursor``, stopping when no cursor comes back. A mirror that + predates windowed pagination ignores the params and returns the full + list with no ``next_cursor`` — the loop then completes in one page, so + this degrades cleanly to a single unbounded request. + """ + merged: Optional[dict] = None + items: list = [] + cursor: Optional[str] = None + for _ in range(MIRROR_MAX_PAGES): + params: dict = {'limit': MIRROR_PAGE_LIMIT} + if cursor: + params['cursor'] = cursor + data = self._request(method, path, params=params, json_body=json_body) + if merged is None: + merged = data + page_items = data.get(list_key) + if page_items: + items.extend(page_items) + cursor = data.get('next_cursor') + if not cursor: + break + else: + bt.logging.warning( + f'Mirror {method} {path} stopped at the {MIRROR_MAX_PAGES}-page cap with a ' + f'cursor still set; the {len(items)} {list_key} loaded may be incomplete.' + ) + if merged is None: + merged = {} + merged[list_key] = items + return merged def _get(self, path: str, params: Optional[dict] = None) -> dict: return self._request('GET', path, params=params) - def _post(self, path: str, json_body: dict) -> dict: - return self._request('POST', path, json_body=json_body) - def _request( self, method: str, @@ -159,7 +208,7 @@ def _request( for attempt in range(self.max_attempts): try: if method == 'POST': - response = self.session.post(url, json=json_body, timeout=self.timeout) + response = self.session.post(url, json=json_body, params=params, timeout=self.timeout) else: response = self.session.get(url, params=params, timeout=self.timeout) except requests.RequestException as e: diff --git a/tests/utils/test_mirror_client.py b/tests/utils/test_mirror_client.py index a6103c0d6..3f81ba577 100644 --- a/tests/utils/test_mirror_client.py +++ b/tests/utils/test_mirror_client.py @@ -27,6 +27,7 @@ MirrorPullRequestsResponse = mirror_models.MirrorPullRequestsResponse MirrorIssuesResponse = mirror_models.MirrorIssuesResponse MirrorPullRequestFilesResponse = mirror_models.MirrorPullRequestFilesResponse +MIRROR_PAGE_LIMIT = pytest.importorskip('gittensor.constants').MIRROR_PAGE_LIMIT # ============================================================================ @@ -88,6 +89,47 @@ def _minimal_files_payload() -> dict: } +def _pr(number: int) -> dict: + """Minimal PR dict that MirrorPullRequest.from_dict parses without warnings.""" + return { + 'repo_full_name': 'o/r', + 'pr_number': number, + 'state': 'OPEN', + 'author_github_id': '218712309', + 'created_at': '2026-03-15T00:00:00Z', + 'review_summary': {'maintainer_changes_requested_count': 0}, + } + + +def _issue(number: int) -> dict: + """Minimal issue dict that MirrorIssue.from_dict parses cleanly.""" + return { + 'repo_full_name': 'o/r', + 'issue_number': number, + 'state': 'OPEN', + } + + +def _pulls_page(prs: list, next_cursor) -> dict: + return { + 'github_id': '218712309', + 'since': None, + 'generated_at': '2026-04-21T00:00:00Z', + 'pull_requests': prs, + 'next_cursor': next_cursor, + } + + +def _issues_page(issues: list, next_cursor) -> dict: + return { + 'github_id': '218712309', + 'since': None, + 'generated_at': '2026-04-21T00:00:00Z', + 'issues': issues, + 'next_cursor': next_cursor, + } + + # ============================================================================ # URL + param construction # ============================================================================ @@ -428,6 +470,99 @@ def test_post_404_fails_fast_no_retry(self, _log, mock_sleep): mock_sleep.assert_not_called() +# ============================================================================ +# Pagination (cursor following) +# ============================================================================ + + +class TestPagination: + """The miner list endpoints page through ``next_cursor`` and concatenate + each page's rows; ``limit`` rides every request, ``cursor`` every request + after the first.""" + + def test_get_pulls_follows_next_cursor_across_pages(self): + session = Mock() + session.get.side_effect = [ + _ok(_pulls_page([_pr(1), _pr(2)], 'CURSOR2')), + _ok(_pulls_page([_pr(3)], None)), + ] + client = _make_client(session) + + result = client.get_miner_pulls('218712309') + + assert session.get.call_count == 2 + assert session.get.call_args_list[0].kwargs['params'] == {'limit': MIRROR_PAGE_LIMIT} + assert session.get.call_args_list[1].kwargs['params'] == {'limit': MIRROR_PAGE_LIMIT, 'cursor': 'CURSOR2'} + assert [pr.pr_number for pr in result.pull_requests] == [1, 2, 3] + + def test_windowed_post_follows_next_cursor_and_keeps_body(self): + session = Mock() + session.post.side_effect = [ + _ok(_pulls_page([_pr(1)], 'CURSOR2')), + _ok(_pulls_page([_pr(2)], None)), + ] + client = _make_client(session) + + result = client.get_miner_pulls('218712309', since_by_repo={'o/r': datetime(2026, 3, 1, tzinfo=timezone.utc)}) + + assert session.post.call_count == 2 + session.get.assert_not_called() + assert session.post.call_args_list[0].kwargs['params'] == {'limit': MIRROR_PAGE_LIMIT} + assert session.post.call_args_list[1].kwargs['params'] == {'limit': MIRROR_PAGE_LIMIT, 'cursor': 'CURSOR2'} + # The per-repo window body rides along on every page. + assert 'since_by_repo' in session.post.call_args_list[1].kwargs['json'] + assert [pr.pr_number for pr in result.pull_requests] == [1, 2] + + def test_get_issues_follows_next_cursor_across_pages(self): + session = Mock() + session.get.side_effect = [ + _ok(_issues_page([_issue(1)], 'C')), + _ok(_issues_page([_issue(2)], None)), + ] + client = _make_client(session) + + result = client.get_miner_issues('218712309') + + assert session.get.call_count == 2 + assert [i.issue_number for i in result.issues] == [1, 2] + + def test_missing_next_cursor_stops_after_one_page(self): + """An un-upgraded mirror returns the full list with no next_cursor — the + loop must complete in a single request (clean degradation).""" + session = Mock() + session.get.return_value = _ok(_minimal_pulls_payload()) + client = _make_client(session) + + result = client.get_miner_pulls('218712309') + + assert session.get.call_count == 1 + assert result.pull_requests == [] + + def test_null_next_cursor_stops_paging(self): + session = Mock() + session.get.return_value = _ok(_pulls_page([_pr(1)], None)) + client = _make_client(session) + + result = client.get_miner_pulls('218712309') + + assert session.get.call_count == 1 + assert [pr.pr_number for pr in result.pull_requests] == [1] + + @patch('gittensor.utils.mirror.client.bt.logging') + @patch('gittensor.utils.mirror.client.MIRROR_MAX_PAGES', 3) + def test_page_cap_halts_runaway_pagination_and_warns(self, mock_log): + session = Mock() + # Every page hands back a cursor — without the cap this never terminates. + session.get.return_value = _ok(_pulls_page([_pr(1)], 'NEVER_ENDS')) + client = _make_client(session) + + result = client.get_miner_pulls('218712309') + + assert session.get.call_count == 3 + assert len(result.pull_requests) == 3 + mock_log.warning.assert_called_once() + + # ============================================================================ # Constructor defaults # ============================================================================