fix(source-github): prevent blocking sleep during rate limit handling (#74758)

Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Co-authored-by: bot_apk <apk@cognition.ai>
Co-authored-by: Serhii Lazebnyi <gl_serhii.lazebnyi@airbyte.io>
Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
Co-authored-by: Serhii Lazebnyi <serglazebny@gmail.com>
Co-authored-by: Daryna Ishchenko <darina.ishchenko17@gmail.com>

Author: 5 people

airbyte-integrations/connectors/source-github/metadata.yaml

@@ -10,7 +10,7 @@ data:
   connectorSubtype: api
   connectorType: source
   definitionId: ef69ef6e-aa7f-4af1-a01d-ef775033524e
-  dockerImageTag: 2.1.21
+  dockerImageTag: 2.1.22
   dockerRepository: airbyte/source-github
   documentationUrl: https://docs.airbyte.com/integrations/sources/github
   erdUrl: https://dbdocs.io/airbyteio/source-github?view=relationships

airbyte-integrations/connectors/source-github/pyproject.toml

@@ -3,7 +3,7 @@ requires = [ "poetry-core>=1.0.0",]
 build-backend = "poetry.core.masonry.api"
 
 [tool.poetry]
-version = "2.1.21"
+version = "2.1.22"
 name = "source-github"
 description = "Source implementation for GitHub."
 authors = [ "Airbyte <contact@airbyte.io>",]

airbyte-integrations/connectors/source-github/source_github/source.py

@@ -241,7 +241,7 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         # This parameter is deprecated and in future will be used sane default, page_size: 10
         page_size = config.get("page_size_for_large_streams", constants.DEFAULT_PAGE_SIZE_FOR_LARGE_STREAM)
         access_token_type, _ = self.get_access_token(config)
-        max_waiting_time = config.get("max_waiting_time", 10) * 60
+        max_waiting_time = config.get("max_waiting_time", 120) * 60
         organization_args = {
             "authenticator": authenticator,
             "organizations": organizations,

airbyte-integrations/connectors/source-github/source_github/spec.json

@@ -135,11 +135,11 @@
       "max_waiting_time": {
         "type": "integer",
         "title": "Max Waiting Time (in minutes)",
-        "examples": [10, 30, 60],
-        "default": 10,
+        "examples": [10, 60, 120],
+        "default": 120,
         "minimum": 1,
-        "maximum": 60,
-        "description": "Max Waiting Time for rate limit. Set higher value to wait till rate limits will be resetted to continue sync",
+        "maximum": 240,
+        "description": "Max time (in minutes) the connector will wait when all API tokens are rate-limited before failing. GitHub rate limits reset every 60 minutes, so values above 60 allow the connector to wait for a full reset cycle.",
         "order": 5
       }
     }

airbyte-integrations/connectors/source-github/source_github/streams.py

@@ -190,11 +190,11 @@ def read_records(self, stream_slice: Mapping[str, Any] = None, **kwargs) -> Iter
 
             self.logger.warning(error_msg)
         except GitHubAPILimitException as e:
-            internal_message = (
-                f"Stream: `{self.name}`, slice: `{stream_slice}`. Limits for all provided tokens are reached, please try again later"
-            )
-            message = "Rate Limits for all provided tokens are reached. For more information please refer to documentation: https://docs.airbyte.com/integrations/sources/github#limitations--troubleshooting"
-            raise AirbyteTracedException(internal_message=internal_message, message=message, failure_type=FailureType.config_error) from e
+            internal_message = f"Stream: `{self.name}`, slice: `{stream_slice}`. {e}"
+            message = "Rate limit exceeded for all configured GitHub API tokens."
+            raise AirbyteTracedException(
+                internal_message=internal_message, message=message, failure_type=FailureType.transient_error
+            ) from e
 
 
 class GithubStream(GithubStreamABC):

airbyte-integrations/connectors/source-github/source_github/utils.py

@@ -56,9 +56,16 @@ class MultipleTokenAuthenticatorWithRateLimiter(AbstractHeaderAuthenticator):
     If a token exceeds the capacity limit, the system switches to another token.
     If all tokens are exhausted, the system will enter a sleep state until
     the first token becomes available again.
+
+    An API budget mechanism throttles requests proactively: when a token's
+    remaining quota drops below a configurable reserve, a small delay is
+    injected before the request so that the connector never fully exhausts
+    all tokens at once.
     """
 
     DURATION = timedelta(seconds=3600)  # Duration at which the current rate limit window resets
+    BUDGET_RESERVE_FRACTION = 0.1  # Start throttling when only 10% of quota remains
+    BUDGET_MIN_RESERVE = 50  # Always keep at least this many calls in reserve per token
 
     def __init__(self, tokens: List[str], auth_method: str = "token", auth_header: str = "Authorization"):
         self._logger = logging.getLogger("airbyte")
@@ -72,7 +79,8 @@ def __init__(self, tokens: List[str], auth_method: str = "token", auth_header: s
         self.check_all_tokens()
         self._tokens_iter = cycle(self._tokens)
         self._active_token = next(self._tokens_iter)
-        self._max_time = 60 * 10  # 10 minutes as default
+        self._max_time = 60 * 120  # 120 minutes as default (must exceed GitHub's 60-min rate limit window)
+        self._budget_logged = False  # avoid log spam for throttle messages
 
     def _initialize_http_clients(self, tokens: List[str]) -> Mapping[str, HttpClient]:
         return {
@@ -169,15 +177,94 @@ def _check_token_limits(self, token: str):
     def check_all_tokens(self):
         for token in self._tokens:
             self._check_token_limits(token)
+        self._budget_logged = False
+
+    def _get_budget_reserve(self, token: Token, count_attr: str) -> int:
+        """Return the minimum number of calls to keep in reserve for a token.
+
+        The reserve is the larger of ``BUDGET_MIN_RESERVE`` and
+        ``BUDGET_RESERVE_FRACTION`` of the token's actual remaining count.
+        We use the current remaining value as an approximation when the
+        original limit is unknown.
+        """
+        remaining = getattr(token, count_attr)
+        # Use 5000 (GitHub default) as the basis, but fall back to the
+        # remaining count if it's higher (e.g. enterprise tokens).
+        limit_estimate = max(5000, remaining)
+        return max(self.BUDGET_MIN_RESERVE, int(limit_estimate * self.BUDGET_RESERVE_FRACTION))
+
+    def _apply_budget_throttle(self, token: Token, count_attr: str, reset_attr: str) -> None:
+        """Optionally sleep a little to spread remaining calls over the reset window.
+
+        When the remaining count for *all* tokens is below the budget
+        reserve, we inject a short delay proportional to how much time
+        remains until the rate-limit window resets.  This avoids hitting
+        the wall and having to do a long blocking sleep.
+        """
+        reserve = self._get_budget_reserve(token, count_attr)
+        remaining = getattr(token, count_attr)
+        if remaining > reserve:
+            return  # plenty of headroom — no throttling needed
+
+        # Only throttle when *every* token is running low so that we don't
+        # slow down needlessly while other tokens still have capacity.
+        if not all(getattr(t, count_attr) <= self._get_budget_reserve(t, count_attr) for t in self._tokens.values()):
+            return
+
+        # Calculate a proportional delay: spread the remaining calls evenly
+        # across the time left until the earliest reset.
+        seconds_to_reset = max((getattr(token, reset_attr) - ab_datetime_now()).total_seconds(), 0)
+        total_remaining = sum(max(getattr(t, count_attr), 0) for t in self._tokens.values())
+        if total_remaining <= 0 or seconds_to_reset <= 0:
+            return
+
+        delay = seconds_to_reset / total_remaining
+        # Cap the delay to avoid extremely long pauses on single requests
+        delay = min(delay, 10.0)
+        if delay >= 0.1:
+            if not self._budget_logged:
+                self._logger.info(
+                    "API budget: throttling requests (%.1fs delay). %d calls remaining across %d token(s), " "%.0fs until reset.",
+                    delay,
+                    total_remaining,
+                    len(self._tokens),
+                    seconds_to_reset,
+                )
+                self._budget_logged = True
+            time.sleep(delay)
+
+    HEARTBEAT_INTERVAL = 60.0  # Log every 60s during exhaustion sleep
+
+    def _sleep_with_heartbeat(self, total_seconds: float, count_attr: str) -> None:
+        """Sleep for *total_seconds* but log progress periodically so the
+        platform heartbeat stays alive and operators can see the connector
+        is not stuck."""
+        remaining = total_seconds
+        while remaining > 0:
+            chunk = min(remaining, self.HEARTBEAT_INTERVAL)
+            time.sleep(chunk)
+            remaining -= chunk
+            if remaining > 0:
+                self._logger.info(
+                    "Rate limit exhausted (%s). Waiting for reset — %.0fs remaining.",
+                    count_attr,
+                    remaining,
+                )
 
     def process_token(self, current_token, count_attr, reset_attr):
         if getattr(current_token, count_attr) > 0:
+            self._apply_budget_throttle(current_token, count_attr, reset_attr)
             setattr(current_token, count_attr, getattr(current_token, count_attr) - 1)
             return True
         elif all(getattr(x, count_attr) == 0 for x in self._tokens.values()):
             min_time_to_wait = min((getattr(x, reset_attr) - ab_datetime_now()).total_seconds() for x in self._tokens.values())
             if min_time_to_wait < self.max_time:
-                time.sleep(min_time_to_wait if min_time_to_wait > 0 else 0)
+                self._logger.info(
+                    "All tokens exhausted (%s). Sleeping %.0fs until rate limit resets.",
+                    count_attr,
+                    max(min_time_to_wait, 0),
+                )
+                self._sleep_with_heartbeat(max(min_time_to_wait, 0), count_attr)
                 self.check_all_tokens()
             else:
                 raise GitHubAPILimitException(f"Rate limits for all tokens ({count_attr}) were reached")

airbyte-integrations/connectors/source-github/unit_tests/test_multiple_token_authenticator.py

@@ -43,7 +43,8 @@ def test_authenticator_counter(rate_limit_mock_response, requests_mock):
     assert authenticator._tokens["token1"].count_rest == 4998
 
 
-def test_multiple_token_authenticator_with_rate_limiter(requests_mock):
+@patch("time.sleep")
+def test_multiple_token_authenticator_with_rate_limiter(sleep_mock, requests_mock):
     """
     This test ensures that:
      1. The rate limiter iterates over all tokens one-by-one after the previous is fully drained.
@@ -89,19 +90,17 @@ def request_callback_orgs(request, context):
     with pytest.raises(AirbyteTracedException) as e:
         list(read_full_refresh(stream))
     assert [(x.count_rest, x.count_graphql) for x in authenticator._tokens.values()] == [(0, 500), (0, 500), (0, 500)]
-    message = (
-        "Stream: `organizations`, slice: `{'organization': 'org1'}`. Limits for all provided tokens are reached, please try again later"
-    )
-    assert e.value.failure_type == FailureType.config_error
-    assert e.value.internal_message == message
+    assert e.value.failure_type == FailureType.transient_error
+    assert "Stream: `organizations`" in e.value.internal_message
+    assert "Rate limits for all tokens" in e.value.internal_message
 
 
 @freeze_time("2021-01-01 12:00:00")
 @patch("time.sleep")
 def test_multiple_token_authenticator_with_rate_limiter_and_sleep(sleep_mock, caplog, requests_mock):
     """
     This test ensures that:
-     1. The rate limiter will only wait (sleep) for token availability if the nearest available token appears within 600 seconds (see max_time).
+     1. The rate limiter will only wait (sleep) for token availability if the nearest available token appears within max_time (default 7200 seconds / 120 minutes).
      2. Token Counter is reset to new values after 1500 requests were made and last token is still in use.
     """
 
@@ -147,7 +146,18 @@ def request_callback_orgs(request, context):
     )
 
     list(read_full_refresh(stream))
-    sleep_mock.assert_called_once_with(ACCEPTED_WAITING_TIME_IN_SECONDS)
+    # The exhaustion sleep is now chunked into heartbeat intervals.
+    # Verify total sleep time adds up to the expected wait time.
+    # Budget-throttle sleeps (small fractional values) may also appear.
+    all_sleeps = [c.args[0] for c in sleep_mock.call_args_list]
+    total_sleep = sum(all_sleeps)
+    # Total sleep should be at least the expected waiting time (budget throttle adds small extras)
+    assert (
+        total_sleep >= ACCEPTED_WAITING_TIME_IN_SECONDS
+    ), f"Expected total sleep >= {ACCEPTED_WAITING_TIME_IN_SECONDS}s, got {total_sleep:.1f}s"
+    # Verify heartbeat chunking: there should be multiple sleep calls for the exhaustion wait
+    heartbeat_sleeps = [s for s in all_sleeps if s >= 1.0]
+    assert len(heartbeat_sleeps) > 1, "Expected multiple heartbeat sleep chunks, not a single blocking sleep"
     assert [(x.count_rest, x.count_graphql) for x in authenticator._tokens.values()] == [(500, 500), (500, 500), (498, 500)]
 
 
@@ -167,3 +177,108 @@ def test_invalid_credentials_error_message(requests_mock):
         MultipleTokenAuthenticatorWithRateLimiter(tokens=["token1", "token2", "token3"])
 
     assert "HTTP Status Code: 401" in e.value.message
+
+
+@freeze_time("2021-01-01 12:00:00")
+@patch("time.sleep")
+def test_api_budget_throttles_when_tokens_run_low(sleep_mock, requests_mock):
+    """
+    Verify that the API budget mechanism injects small delays once all
+    tokens drop below the reserve threshold, preventing full exhaustion.
+    """
+    low_remaining = 30  # Below BUDGET_MIN_RESERVE (50)
+    reset_time = int((ab_datetime_now() + timedelta(seconds=300)).timestamp())
+
+    requests_mock.get(
+        "https://api.github.com/rate_limit",
+        json={
+            "resources": {
+                "core": {"limit": 5000, "used": 4970, "remaining": low_remaining, "reset": reset_time},
+                "graphql": {"limit": 5000, "used": 0, "remaining": 5000, "reset": reset_time},
+            }
+        },
+    )
+
+    authenticator = MultipleTokenAuthenticatorWithRateLimiter(tokens=["token1"])
+    token = authenticator._tokens["token1"]
+    assert token.count_rest == low_remaining
+
+    # Simulate a single request — budget throttle should fire because
+    # remaining (30) < BUDGET_MIN_RESERVE (50)
+    requests_mock.get("https://api.github.com/orgs/org1", json={"id": 1})
+    organization_args = {"organizations": ["org1"], "authenticator": authenticator}
+    stream = Organizations(**organization_args)
+    list(read_full_refresh(stream))
+
+    # time.sleep should have been called with a positive delay for throttling
+    assert sleep_mock.call_count >= 1
+    for call in sleep_mock.call_args_list:
+        assert call.args[0] > 0
+
+
+@freeze_time("2021-01-01 12:00:00")
+@patch("time.sleep")
+def test_api_budget_does_not_throttle_with_headroom(sleep_mock, requests_mock):
+    """
+    Verify that no throttling occurs when tokens have plenty of remaining calls.
+    """
+    reset_time = int((ab_datetime_now() + timedelta(seconds=3600)).timestamp())
+
+    requests_mock.get(
+        "https://api.github.com/rate_limit",
+        json={
+            "resources": {
+                "core": {"limit": 5000, "used": 0, "remaining": 5000, "reset": reset_time},
+                "graphql": {"limit": 5000, "used": 0, "remaining": 5000, "reset": reset_time},
+            }
+        },
+    )
+
+    authenticator = MultipleTokenAuthenticatorWithRateLimiter(tokens=["token1"])
+    requests_mock.get("https://api.github.com/orgs/org1", json={"id": 1})
+    organization_args = {"organizations": ["org1"], "authenticator": authenticator}
+    stream = Organizations(**organization_args)
+    list(read_full_refresh(stream))
+
+    # No throttle delay should have been introduced
+    sleep_mock.assert_not_called()
+
+
+@freeze_time("2021-01-01 12:00:00")
+@patch("time.sleep")
+def test_api_budget_no_throttle_when_some_tokens_have_headroom(sleep_mock, requests_mock):
+    """
+    When only some tokens are below the reserve but others are not, no
+    throttling should occur — the connector should just rotate to a
+    healthy token.
+    """
+    reset_time = int((ab_datetime_now() + timedelta(seconds=300)).timestamp())
+    call_count = 0
+
+    def rate_limit_callback(request, context):
+        nonlocal call_count
+        call_count += 1
+        # First token: low remaining. Second token: plenty of headroom.
+        if call_count <= 1:
+            remaining = 20
+        else:
+            remaining = 4000
+        return json.dumps(
+            {
+                "resources": {
+                    "core": {"limit": 5000, "used": 5000 - remaining, "remaining": remaining, "reset": reset_time},
+                    "graphql": {"limit": 5000, "used": 0, "remaining": 5000, "reset": reset_time},
+                }
+            }
+        )
+
+    requests_mock.get("https://api.github.com/rate_limit", text=rate_limit_callback)
+    authenticator = MultipleTokenAuthenticatorWithRateLimiter(tokens=["token_low", "token_high"])
+
+    requests_mock.get("https://api.github.com/orgs/org1", json={"id": 1})
+    organization_args = {"organizations": ["org1"], "authenticator": authenticator}
+    stream = Organizations(**organization_args)
+    list(read_full_refresh(stream))
+
+    # No throttle — the second token still has headroom
+    sleep_mock.assert_not_called()

docs/integrations/sources/github.md

@@ -229,6 +229,7 @@ Your token should have at least the `repo` scope. Depending on which streams you
 
 | Version    | Date       | Pull Request                                                                                                      | Subject                                                                                                                                                                |
 |:-----------|:-----------|:------------------------------------------------------------------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 2.1.22 | 2026-04-22 | [74758](https://github.com/airbytehq/airbyte/pull/74758) | Fix rate limit sleep blocking heartbeat; classify rate limit errors as transient; increase default max_waiting_time to 120 minutes to cover GitHub's hourly rate limit window |
 | 2.1.21 | 2026-04-21 | [74586](https://github.com/airbytehq/airbyte/pull/74586) | Update dependencies |
 | 2.1.20 | 2026-04-13 | [76275](https://github.com/airbytehq/airbyte/pull/76275) | Bump airbyte-cdk to >=7.12.0 to fix OAuth `scope=null` in consent URL |
 | 2.1.19 | 2026-04-06 | [76090](https://github.com/airbytehq/airbyte/pull/76090) | Fix 403 permission errors misclassified as retryable, causing infinite retry loops |