fix(claude-admin): add rate-limit governor to stop the 429 storm #1927
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,5 +1,5 @@ | ||
| name: claude-admin | ||
| version: "2026.07.25" | ||
| schedule: '0 2 * * *' | ||
| dbt_select: 'tag:claude-admin+' | ||
| workflow: sync | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| """Claude Admin connector test config builder.""" | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| from connector_tests import ConfigBuilder | ||
|
|
||
| API_BASE = "https://api.anthropic.com" | ||
|
|
||
|
|
||
| class ClaudeAdminConfigBuilder(ConfigBuilder): | ||
| def __init__(self) -> None: | ||
| super().__init__() | ||
| self._config.update( | ||
| { | ||
| "admin_api_key": "test-admin-key", | ||
| # Full ISO 8601 with time — the messages_usage and cost_report | ||
| # cursors parse strictly and reject bare YYYY-MM-DD. With the clock | ||
| # frozen at 2026-04-27 and step P1D this yields a small, fixed set | ||
| # of one-day slices from 2026-04-24. | ||
| "start_date": "2026-04-24T00:00:00Z", | ||
| } | ||
| ) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,7 @@ | ||
| import sys | ||
| from pathlib import Path | ||
|
|
||
| # Local builder modules (config.py) are importable under --import-mode=importlib. | ||
| sys.path.insert(0, str(Path(__file__).parent)) | ||
|
|
||
| from connector_tests.plugin import * # noqa: E402,F401,F403 — http_mocker fixture + hooks | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| { | ||
| "id": "user_01TESTAAAAAAAAAAAAAAAAAA", | ||
| "type": "user", | ||
| "email": "member@example.com", | ||
| "name": "Test Member", | ||
| "role": "developer", | ||
| "status": "active", | ||
| "added_at": "2026-01-01T00:00:00Z", | ||
| "last_active_at": "2026-04-24T12:00:00Z" | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,107 @@ | ||
| """Rate-limit / error-handler tests for the claude-admin connector (#1902). | ||
|
|
||
| Every stream's requester references the shared `retryable_error_handler` | ||
| (CompositeErrorHandler): RATE_LIMITED on 429 with WaitTimeFromHeader | ||
| `Retry-After`, RETRY on 5xx with exponential backoff, FAIL on 401/404. The | ||
| connector also sets `concurrency_level: 1` so streams read serially against the | ||
| org-wide Admin API limit instead of firing in parallel (the #1902 429 storm). | ||
|
|
||
| These tests exercise the error handler on the `claude_admin_users` stream | ||
| (simple full-refresh, `data` extractor, after_id pagination). A list of | ||
| responses on one matcher is served consecutively, so [429, 200] proves the | ||
| 429 is retried and the read recovers without record loss. | ||
|
|
||
| Coverage matrix rows: error_retry (429 -> recover), error_retry (503 -> recover), | ||
| error_fail (401 surfaces as an error, no partial records). | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import json | ||
|
|
||
| from config import API_BASE, ClaudeAdminConfigBuilder | ||
|
|
||
| from connector_tests import ( | ||
| ANY_QUERY_PARAMS, | ||
| HttpMocker, | ||
| HttpRequest, | ||
| HttpResponse, | ||
| load_fixture, | ||
| read_stream, | ||
| ) | ||
|
|
||
| _STREAM = "claude_admin_users" | ||
| _CONNECTOR = "ai/claude-admin" | ||
| _URL = f"{API_BASE}/v1/organizations/users" | ||
|
|
||
|
|
||
| def _page(users: list[dict]) -> HttpResponse: | ||
| # has_more=false stops the after_id paginator after one page. | ||
| return HttpResponse( | ||
| body=json.dumps({"data": users, "has_more": False, "last_id": None}), | ||
| status_code=200, | ||
| ) | ||
|
|
||
|
|
||
| def _rate_limited() -> HttpResponse: | ||
| # Retry-After: 0 keeps the test fast; WaitTimeFromHeader reads this header. | ||
| return HttpResponse( | ||
| body=json.dumps({"type": "error", "error": {"type": "rate_limit_error", "message": "Too many requests"}}), | ||
| status_code=429, | ||
| headers={"Retry-After": "0"}, | ||
| ) | ||
|
|
||
|
|
||
| def _server_error() -> HttpResponse: | ||
| return HttpResponse( | ||
| body=json.dumps({"type": "error", "error": {"type": "api_error", "message": "temporary"}}), | ||
| status_code=503, | ||
| ) | ||
|
|
||
|
|
||
| def test_error_retry_429_then_recovers(http_mocker: HttpMocker) -> None: | ||
| """A 429 with Retry-After is retried per the handler; the read succeeds | ||
| once the source recovers, with no record loss and no ERROR log.""" | ||
| config = ClaudeAdminConfigBuilder().build() | ||
| http_mocker.get( | ||
| HttpRequest(_URL, query_params=ANY_QUERY_PARAMS), | ||
| [_rate_limited(), _page([load_fixture(__file__, "user.json")])], | ||
| ) | ||
|
|
||
| output = read_stream(_CONNECTOR, _STREAM, config) | ||
|
|
||
| assert len(output.records) == 1 | ||
| assert not output.errors | ||
| assert output.records[0].record.data["email"] == "member@example.com" | ||
|
|
||
|
|
||
| def test_error_retry_503_then_recovers(http_mocker: HttpMocker) -> None: | ||
| """Transient 5xx is retried with exponential backoff and the read recovers.""" | ||
| config = ClaudeAdminConfigBuilder().build() | ||
| http_mocker.get( | ||
| HttpRequest(_URL, query_params=ANY_QUERY_PARAMS), | ||
| [_server_error(), _page([load_fixture(__file__, "user.json")])], | ||
| ) | ||
|
|
||
| output = read_stream(_CONNECTOR, _STREAM, config) | ||
|
|
||
| assert len(output.records) == 1 | ||
| assert not output.errors | ||
|
|
||
|
|
||
| def test_error_fail_401_surfaces_error(http_mocker: HttpMocker) -> None: | ||
| """401 hits the FAIL branch: the read surfaces an error and emits no | ||
| partial records (rather than silently succeeding with 0 rows).""" | ||
| config = ClaudeAdminConfigBuilder().build() | ||
| http_mocker.get( | ||
| HttpRequest(_URL, query_params=ANY_QUERY_PARAMS), | ||
| HttpResponse( | ||
| body=json.dumps({"type": "error", "error": {"type": "authentication_error", "message": "invalid key"}}), | ||
| status_code=401, | ||
| ), | ||
| ) | ||
|
|
||
| output = read_stream(_CONNECTOR, _STREAM, config, expecting_exception=True) | ||
|
|
||
| assert output.errors, "a 401 must surface as a stream error, not a silent 0-row success" | ||
| assert len(output.records) == 0 | ||
|
Comment on lines
+92
to
+107
There was a problem hiding this comment. 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "Locate relevant files"
fd -a 'test_claude_admin_reliability.py|.*claude.*admin.*reliability.*|.*claude.*admin.*\.(py)$' . | sed 's#^\./##' | head -100
echo
echo "Inspect target test file around 60-130"
tgt=$(fd 'test_claude_admin_reliability.py' . | head -1)
echo "target=$tgt"
wc -l "$tgt"
sed -n '1,150p' "$tgt" | cat -n
echo
echo "Search output.errors uses for ClaudeAdmin"
rg -n "read_stream|errors\\[|assert output\.errors|authentication_error|FAILED|FAIL" .Repository: constructorfabric/insight Length of output: 34691 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "Inspect read_stream implementation and output data shape"
wc -l src/ingestion/tests/connectors/connector_tests/source.py
sed -n '1,180p' src/ingestion/tests/connectors/connector_tests/source.py | cat -n
echo
echo "Search connector error handler / response status / failing error message references"
rg -n "CompositeErrorHandler|RetryableErrorHandler|response_action|authentication_error|status_code|FAIL|externalMessage|failureType|errors" src/ingestion/connectors/ai/claude-admin src/ingestion/connectors/ai/claude-enterpriseRepository: constructorfabric/insight Length of output: 6934 🏁 Script executed: #!/bin/bash
set -euo pipefail
python3 - <<'PY'
try:
from airbyte_cdk.models import EntrypointOutput
import inspect
print("EntrypointOutput fields:")
for name in dir(EntrypointOutput):
if not name.startswith("_"):
print(f" {name}")
print("\n__annotations__ / __fields__:")
print(getattr(EntrypointOutput, "__annotations__", None))
print(inspect.signature(EntrypointOutput))
print(inspect.getfile(EntrypointOutput))
print("\nDocstring:")
print(EntrypointOutput.__doc__)
# Inspect the module without importing/requiring a real Connector, if possible.
import json
import pathlib
import yaml
mod = __import__(EntrypointOutput.__module__, fromlist=["_airbyte_cdk"])
file_ = inspect.getfile(mod)
print(f"\nModule file: {file_}")
txt = pathlib.Path(file_).read_text(errors="replace")
for i, line in enumerate(txt.splitlines(), 1):
if any(s in line for s in ["class EntrypointOutput", "errors", "externalMessage", "failureType"]):
print(f"{i}: {line}")
entry = None
for i, line in enumerate(txt.splitlines(), 1):
if "class EntrypointOutput" in line:
entry = i
if entry:
print(f"\nFragment around EntrypointOutput:")
lines = txt.splitlines()
start = max(1, entry-3)
end = min(len(lines), entry+120)
for j in range(start, end+1):
print(f"{j}: {lines[j-1]}")
except Exception as exc:
print(type(exc).__name__, ":", exc)
import traceback
traceback.print_exc()
PYRepository: constructorfabric/insight Length of output: 339 🏁 Script executed: #!/bin/bash
set -euo pipefail
echo "Search EntrypointOutput definitions in installed tools/env if present"
python3 - <<'PY'
import distutils.sysconfig, site, sysconfig, pathlib, os
for p in site.getsitepackages() + [distutils.sysconfig.get_python_lib(), pathlib.Path(sysconfig.get_path("purelib")), pathlib.Path(sysconfig.get_path("platlib"))]:
print(p)
PY
echo
echo "Search for Airbyte entrypoint output structure in test logs/fixtures/helper references"
rg -n "EntrypointOutput|output\.errors\[0\]|errors=" src/ingestion tests scripts packages 2>/dev/null | head -200
echo
echo "Inspect claude-admin manifest error handler and auth setup"
sed -n '1,140p' src/ingestion/connectors/ai/claude-admin/connector.yaml | cat -n
sed -n '1,140p' src/ingestion/connectors/ai/claude-admin/tests/test_claude_admin_reliability.py | cat -nRepository: constructorfabric/insight Length of output: 1710 Make the 401 assertion specific.
🧰 Tools🪛 ast-grep (0.44.1)[info] 97-100: use JsonResponse instead of HttpResponse to send JSON data (http-response-with-json-dumps) [info] 98-98: use jsonify instead of json.dumps for JSON output (use-jsonify) [error] 97-100: Lack of sanitization of user data (http-response-from-request) 🤖 Prompt for AI Agents |
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Weird claude fix, it usually put the same string if fails to properly run code