diff --git a/tap_github/client.py b/tap_github/client.py index cda1aec..c63332b 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -45,6 +45,19 @@ class GitHubRestStream(RESTStream): # Set to True to use cursor-based pagination instead of page-based pagination use_cursor_pagination = False + # Shared by every GitHubRestStream instance/subclass in this process: once + # any stream trips the secondary (frequency-based) rate limit, we throttle + # all subsequent requests - from any stream - instead of resuming at the + # same pace that triggered it. Decays back down after a run of clean + # requests so a brief burst doesn't slow down an otherwise-healthy sync + # for the rest of the run. Reset each time the tap process starts. + _secondary_rate_limit_delay: ClassVar[float] = 0.0 + _consecutive_clean_requests: ClassVar[int] = 0 + _last_request_at: ClassVar[float | None] = None + _MIN_THROTTLE_DELAY: ClassVar[float] = 1.0 # seconds + _MAX_THROTTLE_DELAY: ClassVar[float] = 30.0 # seconds + _DECAY_AFTER_CLEAN_REQUESTS: ClassVar[int] = 25 + _authenticator: GitHubTokenAuthenticator | None = None @property @@ -79,6 +92,72 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: yield from super().get_records(context) + def _throttle_before_request(self) -> None: + """Sleep if we're currently backing off from a secondary rate limit.""" + delay = GitHubRestStream._secondary_rate_limit_delay + last_request_at = GitHubRestStream._last_request_at + GitHubRestStream._last_request_at = time.monotonic() + if delay <= 0 or last_request_at is None: + return + remaining = delay - (time.monotonic() - last_request_at) + if remaining > 0: + time.sleep(remaining) + + def _note_clean_request(self) -> None: + """Gradually undo the throttle once requests stop tripping the limit.""" + if GitHubRestStream._secondary_rate_limit_delay <= 0: + return + GitHubRestStream._consecutive_clean_requests += 1 + if ( + GitHubRestStream._consecutive_clean_requests + >= self._DECAY_AFTER_CLEAN_REQUESTS + ): + GitHubRestStream._consecutive_clean_requests = 0 + halved = GitHubRestStream._secondary_rate_limit_delay / 2 + GitHubRestStream._secondary_rate_limit_delay = ( + halved if halved >= self._MIN_THROTTLE_DELAY else 0.0 + ) + + def _register_secondary_rate_limit_hit(self) -> None: + """Ratchet up the shared inter-request delay after a secondary rate limit. + + A single sleep-and-retry (see `validate_response`) only paces the one + request that got throttled; every other stream keeps firing at the + same rate that tripped the limit in the first place. This applies a + floor to the gap between *all* subsequent requests, tap-wide, until + enough clean requests go by to relax it again. + """ + GitHubRestStream._consecutive_clean_requests = 0 + current = GitHubRestStream._secondary_rate_limit_delay + new_delay = min( + self._MAX_THROTTLE_DELAY, + max(self._MIN_THROTTLE_DELAY, current * 2), + ) + GitHubRestStream._secondary_rate_limit_delay = new_delay + self.logger.warning( + "Secondary (frequency) rate limit hit - throttling all requests " + f"in this run to >= {new_delay:.1f}s apart until things settle down." + ) + + def _request( + self, + prepared_request: requests.PreparedRequest, + context: Context | None, + ) -> requests.Response: + """Pace requests around the parent implementation. + + Args: + prepared_request: The request to send. + context: Stream partition or context dictionary. + + Returns: + The HTTP response. + """ + self._throttle_before_request() + response = super()._request(prepared_request, context) + self._note_clean_request() + return response + def get_next_page_token( self, response: requests.Response, @@ -261,6 +340,7 @@ def validate_response(self, response: requests.Response) -> None: response.status_code == 403 and "secondary rate limit" in str(response.content).lower() ): + self._register_secondary_rate_limit_hit() # Wait about a minute and retry time.sleep(60 + 30 * random.random()) raise RetriableAPIError(msg, response) diff --git a/tap_github/repository_streams.py b/tap_github/repository_streams.py index 5d2cfff..0d2c6c5 100644 --- a/tap_github/repository_streams.py +++ b/tap_github/repository_streams.py @@ -3,6 +3,7 @@ from __future__ import annotations import http +import time from collections import defaultdict from typing import TYPE_CHECKING, Any, ClassVar from urllib.parse import parse_qs, urlparse @@ -272,9 +273,84 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: "name": context["repo"], "id": context["repo_id"], } + elif context is not None and ( + "organizations" in self.config or "searches" in self.config + ): + # `organizations`/`searches` mode sorts by `updated_at`, in a stable, + # endpoint-imposed order that the "since" param can't actually filter + # (neither endpoint supports it). Every run therefore restarts at the + # same front-of-list repos, so once heavy child streams (commits, + # reviews) exhaust the secondary rate limit, repos further down the + # list are never reached. We round-robin instead: resume right after + # the last repo whose children fully synced last run, wrapping + # around at the end. + yield from self._get_records_round_robin(context) else: yield from super().get_records(context) + def _get_records_round_robin(self, context: Context) -> Iterable[dict[str, Any]]: + """Yield this partition's repos starting after the last completed one. + + Traversal order is our own ascending sort by `id` (immutable), not the + API's `updated_at` order. `updated_at` shifts every run as repos see + real activity - the very thing this stream feeds downstream - so + resuming a position in an `updated_at`-sorted list keeps re-deriving a + different list to resume in. It never skips a repo (every repo the + API returns this run is still yielded exactly once), but each reshuffle + can push already-synced repos back in front of the resume point, + forcing a re-sync of their children before genuinely new repos are + reached. Sorting by `id` removes that churn entirely: new repos always + sort to the end (ids only increase), so the round-robin position is + undisturbed by anyone else's activity. + """ + partition_label = context.get("org") or context.get("search_name") or self.name + + all_records = list(super().get_records(context)) + if not all_records: + self.logger.info(f"[{self.name}:{partition_label}] round-robin summary: 0 repos returned by the API.") + return + + all_records.sort(key=lambda record: record["id"]) + + checkpoint = self.get_context_state(context).setdefault( + "round_robin_checkpoint", {} + ) + last_repo_id = checkpoint.get("last_enumerated_repo_id") + + start_index = 0 + if last_repo_id is not None: + repo_ids = [record["id"] for record in all_records] + if last_repo_id in repo_ids: + start_index = (repo_ids.index(last_repo_id) + 1) % len(all_records) + + ordered_records = all_records[start_index:] + all_records[:start_index] + started_at = time.monotonic() + processed = 0 + + # `finally` runs even on a mid-run crash: when an exception from + # child-stream syncing unwinds the outer sync loop, this generator is + # torn down (GeneratorExit thrown at the `yield` below) before the + # process exits, so the summary still gets logged either way. + try: + for record in ordered_records: + yield record + # Reached only once this record's child streams have fully synced + # (singer_sdk syncs children before pulling the next parent record). + checkpoint["last_enumerated_repo_id"] = record["id"] + checkpoint["last_enumerated_repo"] = record["full_name"] # for readability in state.json + processed += 1 + self._write_state_message() + finally: + full_lap = processed >= len(ordered_records) + self.logger.info( + f"[{self.name}:{partition_label}] round-robin summary: " + f"processed {processed}/{len(ordered_records)} repos in " + f"{time.monotonic() - started_at:.1f}s " + f"({'completed a full lap' if full_lap else 'stopped early - crash or run end'}); " + f"next run resumes after {checkpoint.get('last_enumerated_repo', 'n/a')} " + f"(id={checkpoint.get('last_enumerated_repo_id', 'n/a')})." + ) + schema = th.PropertiesList( th.Property("search_name", th.StringType), th.Property("search_query", th.StringType),