fix: BED-10372 bound collection retries and timeouts

README.md

@@ -55,6 +55,27 @@ Both values must be provided together when overriding the GitHub.com defaults.
 Endpoint URLs must use HTTPS and share the same origin. GitHub App
 `credentials.api_uri` may override only the path on that same origin.
 
+### Request timeouts and retries
+
+REST and GraphQL collection requests use a 10-second connection timeout and a
+30-second read timeout. Transient HTTP 5xx errors, connection failures, and
+timeouts share a limit of three attempts (the initial attempt plus two retries).
+Adaptive GraphQL pagination shares that limit across page sizes: HTTP 502/504
+responses trigger smaller-page attempts instead of a new retry budget.
+
+After retries are exhausted, existing resource handlers log the failure and
+continue with unaffected scopes. Already emitted records are retained, but the
+affected scope may be incomplete. Resources that require all pages before
+emitting records retain that behavior.
+
+Workflow collection skips `dynamic/` paths. Missing workflow files are logged
+and skipped without passing failed items to job or step parsing.
+
+Rate-limit responses and GitHub App token refresh retain their existing retry
+handling and do not consume the transient-failure budget. These timeouts are
+not an overall request deadline. App authentication requests and credential
+inventory CSV downloads retain their separate timeout settings.
+
 ### Enterprise SCIM and hybrid correlations
 
 A token with enterprise SCIM access is used to collect both `/scim/v2/enterprises/{enterprise}/Users` and `/scim/v2/enterprises/{enterprise}/Groups`. The collector emits normalized `SCIM_Organization`, `SCIM_User`, and `SCIM_Group` nodes plus `SCIM_Contains`, `SCIM_MemberOf`, and `SCIM_Provisioned` relationships. Install the BloodHound SCIM extension alongside this extension to register the shared SCIM kinds.

src/openhound_github/github_rest_client.py

@@ -1,3 +1,90 @@
+from contextlib import contextmanager
+from contextvars import ContextVar
+from dataclasses import dataclass
+from functools import wraps
+
+from dlt.sources.helpers import requests
+from requests.exceptions import ChunkedEncodingError, ConnectionError, Timeout
+from tenacity import RetryCallState
+from tenacity.stop import stop_base
+
 from openhound_github.github_retry import should_retry_github_response
 
-__all__ = ["should_retry_github_response"]
+__all__ = ["github_http_session", "should_retry_github_response"]
+
+
+@dataclass
+class GraphQLPageRetryBudget:
+    """Share transient failures across page sizes without changing other requests."""
+
+    failures: int = 0
+
+    @property
+    def exhausted(self) -> bool:
+        return self.failures >= 3
+
+    @contextmanager
+    def activate(self):
+        token = _graphql_page_budget.set(self)
+        try:
+            yield
+        finally:
+            _graphql_page_budget.reset(token)
+
+
+_graphql_page_budget: ContextVar[GraphQLPageRetryBudget | None] = ContextVar(
+    "github_graphql_page_budget", default=None
+)
+
+
+class _StopAfterTransientFailures(stop_base):
+    """Count transient failures per send, independently of other retry reasons."""
+
+    def __call__(self, retry_state: RetryCallState) -> bool:
+        outcome = retry_state.outcome
+        if outcome is None:
+            return False
+        exception = outcome.exception()
+        response = None
+        if isinstance(exception, requests.HTTPError):
+            response = exception.response
+        elif exception is None:
+            response = outcome.result()
+        transient = isinstance(exception, (ConnectionError, Timeout, ChunkedEncodingError)) or (
+            response is not None and 500 <= response.status_code < 600
+        )
+        page_budget = _graphql_page_budget.get()
+        if page_budget is not None:
+            page_budget.failures += int(transient)
+            # Return gateway errors immediately so pagination can shrink the page.
+            return page_budget.exhausted or (
+                response is not None and response.status_code in (502, 504)
+            )
+        failures = getattr(retry_state, "_github_transient_failures", 0) + int(transient)
+        setattr(retry_state, "_github_transient_failures", failures)
+        return failures >= 3
+
+
+def github_http_session(retry_condition):
+    """Bound transient failures while retaining DLT's existing retry policy."""
+    session = requests.Client(
+        request_timeout=(10, 30),
+        status_codes=tuple(range(500, 600)),
+        retry_condition=retry_condition,
+    ).session
+    # Keep the configured overall limit for rate limits and credential refresh.
+    # RetryCallState is new for every send, including concurrent requests.
+    retry = session.send.retry
+    retry.stop = _StopAfterTransientFailures() | retry.stop
+    send = session.send
+
+    @wraps(send)
+    def send_with_status_check(request, **kwargs):
+        response = send(request, **kwargs)
+        # RESTClient calls send directly, bypassing Session.request's status check.
+        # Raise only after the retry loop finishes so scope handlers can skip it.
+        response.raise_for_status()
+        return response
+
+    session.send = send_with_status_check
+    return session

src/openhound_github/helpers.py

@@ -14,6 +14,7 @@
 from requests import Request
 
 from openhound_github.auth import GitHubAppInstallationAuth
+from openhound_github.github_rest_client import GraphQLPageRetryBudget
 from openhound_github.github_retry import (
     is_primary_rate_limit_response,
     is_secondary_rate_limit_response,
@@ -230,9 +231,9 @@ def adaptive_graphql_paginate(
 ) -> Iterator[dict[str, Any]]:
     """Yield GraphQL pages while adapting connection size after gateway failures.
 
-    The underlying HTTP session still owns ordinary request retries. This helper
-    handles the remaining terminal 502/504 case by retrying the same cursor with
-    progressively smaller connection sizes.
+    HTTP retries and smaller-page recovery share three transient attempts per
+    page. Gateway errors return immediately so the next attempt can shrink the
+    page; rate-limit and credential-refresh retries retain their existing policy.
     """
     if not page_sizes or any(page_size <= 0 for page_size in page_sizes):
         raise ValueError("page_sizes must contain positive integers")
@@ -270,6 +271,7 @@ def adaptive_graphql_paginate(
         page_data: dict[str, Any] | None = None
         paginator: GraphQLCursorPaginator | None = None
         successful_page_size: int | None = None
+        retry_budget = GraphQLPageRetryBudget()
 
         for index, page_size in enumerate(candidate_sizes):
             request_variables = {
@@ -278,10 +280,11 @@ def adaptive_graphql_paginate(
                 page_size_variable: page_size,
             }
             try:
-                response = client.post(
-                    graphql_path,
-                    json={"query": query, "variables": request_variables},
-                )
+                with retry_budget.activate():
+                    response = client.post(
+                        graphql_path,
+                        json={"query": query, "variables": request_variables},
+                    )
                 response.raise_for_status()
                 response_json = response.json()
                 paginator = GraphQLCursorPaginator(
@@ -303,7 +306,11 @@ def adaptive_graphql_paginate(
                 status_code = _graphql_gateway_status(error)
                 next_page_size = (
                     candidate_sizes[index + 1]
-                    if status_code is not None and index + 1 < len(candidate_sizes)
+                    if (
+                        status_code is not None
+                        and index + 1 < len(candidate_sizes)
+                        and not retry_budget.exhausted
+                    )
                     else None
                 )
                 if next_page_size is not None:

src/openhound_github/resources/organization.py

@@ -6,6 +6,7 @@
 from typing import Any, Iterator
 from urllib.parse import quote
 
+import dlt
 from dlt.sources.helpers import requests
 from dlt.sources.helpers.rest_client.client import RESTClient
 from dlt.sources.helpers.rest_client.paginators import (
@@ -1475,7 +1476,7 @@ def workflows(repo: Repository, ctx: SourceContext):
         Workflow (Workflow): An active workflow record.
     """
 
-    @app.defer
+    @dlt.defer
     def _workflow_file_contents(
         client: RESTClient,
         repo: Repository,
@@ -1487,11 +1488,30 @@ def _workflow_file_contents(
             return None
 
         params = {"ref": repo.default_branch} if repo.default_branch else None
-        response = client.get(
-            f"/repos/{repo.full_name}/contents/{path}", params=params
-        ).json()
+        try:
+            response = client.get(
+                f"/repos/{repo.full_name}/contents/{path}", params=params
+            ).json()
+            content = response.get("content")
+        except Exception as exc:
+            # Return None so DLT drops the failed item. app.defer returns [],
+            # which can reach the job/step transformers as an invalid Workflow.
+            if (
+                isinstance(exc, requests.HTTPError)
+                and exc.response is not None
+                and exc.response.status_code == 404
+            ):
+                logger.warning(
+                    "Skipping workflow '%s' in repository '%s': file unavailable at ref '%s' (HTTP 404)",
+                    path, repo.full_name, repo.default_branch,
+                )
+            else:
+                logger.error(
+                    "Unable to collect workflow '%s' in repository '%s': %s",
+                    path, repo.full_name, exc,
+                )
+            return None
 
-        content = response.get("content")
         if not content:
             return None
 
@@ -1517,6 +1537,9 @@ def _workflow_file_contents(
     ):
         for workflow in page:
             if workflow.get("state") == "active":
+                path = workflow.get("path")
+                if not path or path.startswith("dynamic/"):
+                    continue
                 if workflow_permissions is None:
                     workflow_permissions = _repository_workflow_permissions(
                         ctx, client, repo.full_name

src/openhound_github/source.py

@@ -8,7 +8,6 @@
 import dlt
 from dlt.common.configuration import configspec
 from dlt.common.configuration.specs import CredentialsConfiguration
-from dlt.sources.helpers import requests
 from dlt.sources.helpers.rest_client.auth import AuthConfigBase
 from dlt.sources.helpers.rest_client.auth import BearerTokenAuth
 from dlt.sources.helpers.rest_client.client import RESTClient
@@ -27,6 +26,7 @@
     DEFAULT_GITHUB_REST_API_URL,
     github_retry_policy,
 )
+from openhound_github.github_rest_client import github_http_session
 from openhound_github.main import app
 from openhound_github.models.saml_helpers import (
     DEFAULT_GITHUB_DEPLOYMENT_ID,
@@ -352,10 +352,7 @@ def api_client(base_url: str, auth: AuthConfigBase) -> RESTClient:
             },
             auth=auth,
             paginator=HeaderLinkPaginator(),
-            session=requests.Client(
-                status_codes=tuple(range(500, 600)),
-                retry_condition=github_retry_policy(auth),
-            ).session,
+            session=github_http_session(github_retry_policy(auth)),
         )
 
     def clients(auth: AuthConfigBase) -> tuple[RESTClient, RESTClient]: