feat: round-robin repo-enumeration checkpoint + secondary rate-limit throttle

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)

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),