From 0c298cad971411971a032ccc5bac03617f87f36b Mon Sep 17 00:00:00 2001 From: Jared Atkinson Date: Thu, 29 Oct 2026 21:29:37 -0700 Subject: [PATCH 1/2] fix: BED-10372 bound collection retries and timeouts --- README.md | 18 +++ src/openhound_github/github_rest_client.py | 89 +++++++++++- src/openhound_github/helpers.py | 23 ++-- src/openhound_github/source.py | 7 +- tests/test_collection_continuation.py | 153 +++++++++++++++++++++ tests/test_github_app_retry.py | 36 +++++ tests/test_github_rest_client.py | 113 ++++++++++++++- tests/test_graphql_retry_budget.py | 100 ++++++++++++++ 8 files changed, 523 insertions(+), 16 deletions(-) create mode 100644 tests/test_collection_continuation.py create mode 100644 tests/test_graphql_retry_budget.py diff --git a/README.md b/README.md index db260df..efe3427 100644 --- a/README.md +++ b/README.md @@ -55,6 +55,24 @@ 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. + +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. diff --git a/src/openhound_github/github_rest_client.py b/src/openhound_github/github_rest_client.py index 7adca6f..38b6c27 100644 --- a/src/openhound_github/github_rest_client.py +++ b/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 diff --git a/src/openhound_github/helpers.py b/src/openhound_github/helpers.py index 076c6bf..ed68314 100644 --- a/src/openhound_github/helpers.py +++ b/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: diff --git a/src/openhound_github/source.py b/src/openhound_github/source.py index 7baebe6..13382fa 100644 --- a/src/openhound_github/source.py +++ b/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]: diff --git a/tests/test_collection_continuation.py b/tests/test_collection_continuation.py new file mode 100644 index 0000000..3116ba0 --- /dev/null +++ b/tests/test_collection_continuation.py @@ -0,0 +1,153 @@ +"""Exercise exhausted retries through the real parallel DLT resource wrappers.""" + +import json +import logging +from collections import Counter +from threading import Lock +from types import SimpleNamespace +from urllib.parse import parse_qs, urlparse + +import dlt +import pytest +from dlt.sources.helpers.rest_client.auth import BearerTokenAuth +from dlt.sources.helpers.rest_client.client import RESTClient +from dlt.sources.helpers.rest_client.paginators import HeaderLinkPaginator +from requests import Response +from requests.exceptions import ReadTimeout + +from openhound_github.github_rest_client import github_http_session +from openhound_github.helpers import github_retry_policy +from openhound_github.resources.organization import ( + OrgContext, + SourceContext, + environment_branch_policies, + environment_secrets, + environment_variables, + environments, + organizations, +) + + +def _client(monkeypatch, route): + session = github_http_session(github_retry_policy(BearerTokenAuth("test"))) + session.send.retry.sleep = lambda _: None + calls = Counter() + lock = Lock() + + def send(request, **kwargs): + url = urlparse(request.url) + page = parse_qs(url.query).get("page", ["1"])[0] + with lock: + calls[url.path, page] += 1 + outcome = route(url.path, page) + if isinstance(outcome, Exception): + raise outcome + status, payload, more = outcome + response = Response() + response.status_code = status + response.url = request.url + response.request = request + response._content = json.dumps(payload).encode() + if more: + response.headers["Link"] = f'; rel="next"' + return response + + monkeypatch.setattr(session.get_adapter("https://"), "send", send) + return RESTClient( + "https://api.github.com", session=session, paginator=HeaderLinkPaginator(), + ), calls + + +def _failure(error): + return ReadTimeout("simulated read timeout") if error == "timeout" else ( + 503, {"message": "simulated unavailable endpoint"}, False, + ) + + +def _field(row, name): + return row[name] if isinstance(row, dict) else getattr(row, name) + + +@pytest.mark.parametrize("error", ["503", "timeout"]) +def test_later_organizations_collect_after_exhausted_retries(monkeypatch, caplog, error): + def route(path, page): + if path == "/orgs/bad": + return _failure(error) + org = path.split("/")[2] + return 200, {"login": org, "node_id": f"ORG_{org}"}, False + + client, calls = _client(monkeypatch, route) + ctx = SourceContext(client=client, organizations=[ + OrgContext(client=client, org_name=name) for name in ("first", "bad", "last") + ]) + + with caplog.at_level(logging.ERROR): + rows = list(organizations(ctx)) + + assert {_field(row, "login") for row in rows} == {"first", "last"} + assert calls["/orgs/bad", "1"] == 3 + assert "processing organization 'bad'" in caplog.text + + +@pytest.mark.parametrize("error", ["503", "timeout"]) +def test_later_repositories_collect_and_partial_page_is_retained(monkeypatch, caplog, error): + def route(path, page): + repo = path.split("/")[3] + if repo == "bad" and page == "2": + return _failure(error) + environment = { + "id": 1, "node_id": f"ENV_{repo}", "name": repo, + "url": "https://api.github.com/example", "html_url": "https://github.com/example", + "created_at": "2026-01-01T00:00:00Z", "updated_at": "2026-01-01T00:00:00Z", + } + return 200, {"environments": [environment]}, repo == "bad" + + client, calls = _client(monkeypatch, route) + ctx = SourceContext(client=client, organizations=[OrgContext(client=client, org_name="org")]) + parents = dlt.resource([ + SimpleNamespace(id=f"REPO_{name}", name=name, org_login="org", environment_count=1) + for name in ("first", "bad", "last") + ], name="probe_repositories") + + with caplog.at_level(logging.ERROR): + rows = list(parents | environments(ctx)) + + assert {_field(row, "repository_name") for row in rows} == {"first", "bad", "last"} + assert calls["/repos/org/bad/environments", "2"] == 3 + assert "resource 'environments'" in caplog.text + assert "/repos/org/bad/environments" in caplog.text or error == "timeout" + + +@pytest.mark.parametrize("transformer, suffix, selector", [ + (environment_variables, "variables", "variables"), + (environment_secrets, "secrets", "secrets"), + (environment_branch_policies, "deployment-branch-policies", "branch_policies"), +]) +@pytest.mark.parametrize("error", ["503", "timeout"]) +def test_later_environments_collect_after_partial_failure( + monkeypatch, caplog, transformer, suffix, selector, error, +): + def route(path, page): + name = path.split("/")[5] + if name == "bad" and page == "2": + return _failure(error) + item = {"name": "item", "value": "value", "id": 1, "node_id": "POLICY", + "created_at": "2026-01-01T00:00:00Z"} + return 200, {selector: [item]}, name == "bad" + + client, calls = _client(monkeypatch, route) + ctx = SourceContext(client=client, organizations=[OrgContext(client=client, org_name="org")]) + parents = dlt.resource([ + SimpleNamespace( + name=name, node_id=f"ENV_{name}", org_login="org", repository_name="repo", + repository_full_name="org/repo", repository_node_id="REPO", + has_custom_branch_policies=True, required_reviewers=False, prevent_self_review=False, + ) for name in ("first", "bad", "last") + ], name="probe_environments") + + with caplog.at_level(logging.ERROR): + rows = list(parents | transformer(ctx)) + + assert {_field(row, "environment_name") for row in rows} == {"first", "bad", "last"} + assert calls[f"/repos/org/repo/environments/bad/{suffix}", "2"] == 3 + assert f"resource '{transformer.name}'" in caplog.text diff --git a/tests/test_github_app_retry.py b/tests/test_github_app_retry.py index 5b9b1fd..cfcbaca 100644 --- a/tests/test_github_app_retry.py +++ b/tests/test_github_app_retry.py @@ -4,10 +4,12 @@ from threading import Event, Lock import requests +import pytest from dlt.sources.helpers.rest_client.auth import BearerTokenAuth from openhound_github.auth import GitHubAppInstallationAuth, TokenResponse from openhound_github.helpers import github_retry_policy +from openhound_github.github_rest_client import github_http_session class FakeInstallation: @@ -243,3 +245,37 @@ def test_retry_policy_does_not_repair_bad_credentials_for_bearer_token_auth() -> assert should_retry is False assert request.headers["Authorization"] == "Bearer static-token" + + +@pytest.mark.parametrize("replacement_rejected", [False, True]) +def test_http_session_retries_with_refreshed_credentials(monkeypatch, replacement_rejected): + installation = FakeInstallation("new-token") + auth = GitHubAppInstallationAuth(installation=installation) + auth.access_token = "old-token" + auth.expires_at = datetime.now(timezone.utc) + timedelta(hours=1) + session = github_http_session(github_retry_policy(auth)) + session.auth = auth + session.send.retry.sleep = lambda _: None + # Two transient failures must leave room for credential recovery. + statuses = iter([503, 503, 401, 401 if replacement_rejected else 200]) + authorizations = [] + + def send(request, **kwargs): + authorizations.append(request.headers.get("Authorization")) + response = requests.Response() + response.status_code = next(statuses) + response.request = request + response.url = request.url + response._content = b'{"message":"Bad credentials"}' + return response + + monkeypatch.setattr(session.get_adapter("https://"), "send", send) + if replacement_rejected: + with pytest.raises(requests.HTTPError) as caught: + session.get("https://api.github.com/example") + assert caught.value.response.status_code == 401 + else: + assert session.get("https://api.github.com/example").status_code == 200 + + assert authorizations == ["Bearer old-token"] * 3 + ["Bearer new-token"] + assert installation.token_calls == 1 diff --git a/tests/test_github_rest_client.py b/tests/test_github_rest_client.py index 1a80551..508303c 100644 --- a/tests/test_github_rest_client.py +++ b/tests/test_github_rest_client.py @@ -1,8 +1,14 @@ import json +from unittest.mock import Mock -from requests import Response +import pytest +from dlt.sources.helpers import requests as dlt_requests +from requests import HTTPError, Response +from requests.exceptions import ChunkedEncodingError, ConnectionError, ReadTimeout +from dlt.sources.helpers.rest_client.auth import BearerTokenAuth -from openhound_github.github_rest_client import should_retry_github_response +from openhound_github.github_rest_client import github_http_session, should_retry_github_response +from openhound_github.helpers import github_retry_policy def make_response( @@ -50,3 +56,106 @@ def test_does_not_retry_non_rate_limit_403() -> None: ) assert should_retry_github_response(response, None) is False + + +def _session(monkeypatch, outcomes): + session = github_http_session(github_retry_policy(BearerTokenAuth("test"))) + send = Mock(side_effect=outcomes) + monkeypatch.setattr(session.get_adapter("https://"), "send", send) + session.send.retry.sleep = lambda _: None + return session, send + + +@pytest.mark.parametrize("status", [500, 502, 503, 504]) +def test_server_failures_stop_after_three_attempts(monkeypatch, status): + session, send = _session(monkeypatch, [make_response(status) for _ in range(4)]) + + with pytest.raises(HTTPError) as caught: + session.get("https://api.github.com/example") + + assert caught.value.response.status_code == status + assert send.call_count == 3 + assert all(call.kwargs["timeout"] == (10, 30) for call in send.call_args_list) + + +@pytest.mark.parametrize("error", [ReadTimeout, ConnectionError, ChunkedEncodingError]) +def test_network_failures_stop_after_three_attempts(monkeypatch, error): + session, send = _session(monkeypatch, [error("interrupted") for _ in range(4)]) + + with pytest.raises(error): + session.get("https://api.github.com/example") + + assert send.call_count == 3 + + +def test_transient_failure_count_resets_for_each_request(monkeypatch): + session, send = _session(monkeypatch, [ + make_response(503), make_response(503), make_response(200), + make_response(503), make_response(503), make_response(200), + ]) + + assert session.get("https://api.github.com/example").status_code == 200 + assert session.get("https://api.github.com/example").status_code == 200 + assert send.call_count == 6 + + +def test_mixed_server_and_network_failures_share_budget(monkeypatch): + session, send = _session(monkeypatch, [ + make_response(503), ReadTimeout("interrupted"), make_response(502), + make_response(200), + ]) + + with pytest.raises(HTTPError): + session.get("https://api.github.com/example") + + assert send.call_count == 3 + + +def test_rate_limit_retries_do_not_consume_transient_budget(monkeypatch): + session, send = _session(monkeypatch, [ + *[make_response(429, body={"message": "secondary rate limit"}) for _ in range(4)], + make_response(503), make_response(503), make_response(200), + ]) + + assert session.get("https://api.github.com/example").status_code == 200 + assert send.call_count == 7 + + +@pytest.mark.parametrize(("status", "headers", "body"), [ + (403, {"x-ratelimit-remaining": "0", "x-ratelimit-reset": "4102444800"}, + {"message": "API rate limit exceeded"}), + (429, {}, {"message": "secondary rate limit"}), + (429, {"Retry-After": "12"}, {"message": "Please wait"}), + (200, {"x-ratelimit-resource": "graphql", "x-ratelimit-remaining": "0", + "x-ratelimit-reset": "4102444800"}, + {"errors": [{"message": "API rate limit exceeded"}]}), +]) +def test_rate_limit_attempts_and_waits_match_existing_session( + monkeypatch, status, headers, body, +): + monkeypatch.setenv("RUNTIME__REQUEST_MAX_ATTEMPTS", "5") + observations = [] + for updated in (False, True): + policy = github_retry_policy(BearerTokenAuth("test")) + session = ( + github_http_session(policy) if updated else dlt_requests.Client( + status_codes=tuple(range(500, 600)), retry_condition=policy, + ).session + ) + send = Mock(side_effect=[ + make_response(status, headers=headers, body=body) for _ in range(5) + ]) + monkeypatch.setattr(session.get_adapter("https://"), "send", send) + sleep = Mock() + session.send.retry.sleep = sleep + if status != 200: + with pytest.raises(HTTPError): + session.get("https://api.github.com/example") + else: + response = session.get("https://api.github.com/graphql") + assert response.json()["errors"] + observations.append((send.call_count, [float(c.args[0]) for c in sleep.call_args_list])) + + assert observations[0][0] == 5 + assert len(observations[0][1]) == 4 + assert observations[1] == observations[0] diff --git a/tests/test_graphql_retry_budget.py b/tests/test_graphql_retry_budget.py new file mode 100644 index 0000000..834ed3d --- /dev/null +++ b/tests/test_graphql_retry_budget.py @@ -0,0 +1,100 @@ +import json + +import pytest +from dlt.sources.helpers.rest_client.auth import BearerTokenAuth +from dlt.sources.helpers.rest_client.client import RESTClient +from requests import Response +from requests.exceptions import ReadTimeout + +from openhound_github.github_rest_client import github_http_session +from openhound_github.helpers import AdaptiveGraphQLPageError, adaptive_graphql_paginate, github_retry_policy + + +def _page(cursor=None): + return {"data": {"items": {"pageInfo": { + "hasNextPage": cursor is not None, "endCursor": cursor, + }}}} + + +def _client(monkeypatch, outcomes): + auth = BearerTokenAuth("test") + session = github_http_session(github_retry_policy(auth)) + session.send.retry.sleep = lambda _: None + outcomes = iter(outcomes) + calls = [] + + def send(request, **kwargs): + calls.append(json.loads(request.body)["variables"] if request.body else {}) + outcome = next(outcomes) + if isinstance(outcome, Exception): + raise outcome + response = Response() + response.request = request + response.url = request.url + response.status_code = outcome if isinstance(outcome, int) else 200 + payload = outcome if isinstance(outcome, dict) else {"message": "secondary rate limit"} + response._content = json.dumps(payload).encode() + return response + + monkeypatch.setattr(session.get_adapter("https://"), "send", send) + return RESTClient("https://api.github.com", session=session, auth=auth), calls + + +def _pages(client): + return adaptive_graphql_paginate( + client, query="query", variables={"after": None}, + page_info_path="data.items.pageInfo", + ) + + +@pytest.mark.parametrize(("outcomes", "sizes"), [ + ([502, 504, 502], [100, 50, 25]), + ([503, 502, 504], [100, 100, 50]), + ([503, 503, 502], [100, 100, 100]), + ([502, ReadTimeout("interrupted"), 503], [100, 50, 50]), + ([ReadTimeout("interrupted") for _ in range(3)], [100, 100, 100]), +]) +def test_adaptive_page_exhausts_three_shared_attempts(monkeypatch, outcomes, sizes): + client, calls = _client(monkeypatch, [*outcomes, _page()]) + + with pytest.raises(AdaptiveGraphQLPageError) as caught: + list(_pages(client)) + + assert [call["count"] for call in calls] == sizes + assert caught.value.cursor is None + assert caught.value.page_size == sizes[-1] + + +def test_smaller_page_recovers_and_next_page_has_fresh_budget(monkeypatch): + client, calls = _client(monkeypatch, [ + 502, 504, _page("next"), 503, 503, _page(), + ]) + + assert len(list(_pages(client))) == 2 + assert [(call["after"], call["count"]) for call in calls] == [ + (None, 100), (None, 50), (None, 25), + ("next", 100), ("next", 100), ("next", 100), + ] + + +def test_later_failure_retains_earlier_page_and_restores_normal_http_retries(monkeypatch): + client, calls = _client(monkeypatch, [ + _page("next"), 502, 504, 502, 502, 502, _page(), + ]) + pages = _pages(client) + assert next(pages) == _page("next")["data"] + with pytest.raises(AdaptiveGraphQLPageError) as caught: + next(pages) + assert caught.value.cursor == "next" + assert len(calls) == 4 + + # A request outside pagination must still retry gateway failures normally. + assert client.get("/example").status_code == 200 + assert len(calls) == 7 + + +def test_rate_limits_do_not_consume_adaptive_transient_budget(monkeypatch): + client, calls = _client(monkeypatch, [429, 429, 429, 429, 502, 504, _page()]) + + assert len(list(_pages(client))) == 1 + assert [call["count"] for call in calls] == [100] * 5 + [50, 25] From 6c08409e0190f1d166865ea3f6e9e805cd348b62 Mon Sep 17 00:00:00 2001 From: Jared Atkinson Date: Thu, 29 Oct 2026 22:10:38 -0700 Subject: [PATCH 2/2] fix: BED-10372 prevent workflow failure cascades --- README.md | 3 + .../resources/organization.py | 33 ++++++++-- tests/test_workflow_resources.py | 63 ++++++++++++++++++- 3 files changed, 92 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index efe3427..8287fdf 100644 --- a/README.md +++ b/README.md @@ -68,6 +68,9 @@ 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 diff --git a/src/openhound_github/resources/organization.py b/src/openhound_github/resources/organization.py index 31580aa..a0e273e 100644 --- a/src/openhound_github/resources/organization.py +++ b/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 diff --git a/tests/test_workflow_resources.py b/tests/test_workflow_resources.py index 5e100df..a744f85 100644 --- a/tests/test_workflow_resources.py +++ b/tests/test_workflow_resources.py @@ -1,11 +1,16 @@ import inspect +import base64 +import logging from types import SimpleNamespace +import dlt import pytest from dlt.sources.helpers import requests from requests.exceptions import JSONDecodeError -from openhound_github.resources.organization import OrgContext, SourceContext, workflows +from openhound_github.resources.organization import ( + OrgContext, SourceContext, workflows, workflow_jobs, workflow_steps, +) class _FakeResponse: @@ -25,12 +30,14 @@ def __init__( can_approve_pull_request_reviews: bool = False, workflow_permission_responses: dict[str, dict] | None = None, workflow_permission_errors: dict[str, BaseException] | None = None, + workflow_file_errors: dict[str, BaseException] | None = None, ): self.workflow_pages = workflow_pages self.default_workflow_permissions = default_workflow_permissions self.can_approve_pull_request_reviews = can_approve_pull_request_reviews self.workflow_permission_responses = workflow_permission_responses or {} self.workflow_permission_errors = workflow_permission_errors or {} + self.workflow_file_errors = workflow_file_errors or {} self.get_calls: list[tuple[str, dict]] = [] self.paginate_calls: list[tuple[str, dict]] = [] @@ -51,7 +58,10 @@ def get(self, path: str, **kwargs): "can_approve_pull_request_reviews": self.can_approve_pull_request_reviews, } ) - return _FakeResponse({"content": "am9iczoge30="}) + if path in self.workflow_file_errors: + raise self.workflow_file_errors[path] + contents = "jobs:\n build:\n runs-on: ubuntu-latest\n steps:\n - run: echo hello\n" + return _FakeResponse({"content": base64.b64encode(contents.encode()).decode()}) def _repo() -> SimpleNamespace: @@ -270,3 +280,52 @@ def test_workflows_propagate_unexpected_repository_permission_lookup_errors() -> with pytest.raises(RuntimeError, match="unexpected workflow permission failure"): _collect_workflows(_repo(), _ctx(client)) + + +@pytest.mark.parametrize("path", [ + "dynamic/dependabot/update-graph", "dynamic/pages/pages-build-deployment", "", None, +]) +def test_workflows_skip_non_file_paths_before_permission_lookup(path): + workflow = {**_workflow_row(1), "path": path} + client = _FakeClient([[workflow]]) + + assert _collect_workflows(_repo(), _ctx(client)) == [] + assert client.get_calls == [] + + +@pytest.mark.parametrize("child", [workflow_jobs, workflow_steps]) +@pytest.mark.parametrize("failure", ["404", "503", "timeout", "unexpected"]) +def test_failed_workflow_file_does_not_cascade_to_children(caplog, child, failure): + if failure in {"404", "503"}: + response = requests.Response() + response.status_code = int(failure) + error = requests.HTTPError("file fetch failed", response=response) + elif failure == "timeout": + error = requests.ReadTimeout("file fetch timed out") + else: + error = RuntimeError("unexpected fetch failure") + dynamic = {**_workflow_row(4), "path": "dynamic/dependabot/update-graph"} + client = _FakeClient( + [[_workflow_row(1), _workflow_row(2), dynamic, _workflow_row(3)]], + workflow_file_errors={"/repos/acme/repo/contents/.github/workflows/2.yml": error}, + ) + parent = dlt.resource([_repo()], name="probe_workflow_repository") + + with caplog.at_level(logging.WARNING): + rows = list(parent | workflows(_ctx(client)) | child()) + + assert len(rows) == 2 + assert {_field(row, "workflow_node_id") for row in rows} == {"W_1", "W_3"} + assert not any("/contents/dynamic/" in path for path, _ in client.get_calls) + assert "has no attribute" not in caplog.text + assert "Error executing DLT defer" not in caplog.text + assert "2.yml" in caplog.text + if failure == "404": + assert "HTTP 404" in caplog.text + assert not any(record.levelno >= logging.ERROR for record in caplog.records) + else: + assert "Unable to collect workflow" in caplog.text + + +def _field(row, name): + return row[name] if isinstance(row, dict) else getattr(row, name)