fix(oauth): bound token refresh retries for REFRESH_TOKEN_THEN_RETRY

[devin-ai-integration]

Summary

Requested by Daryna Ishchenko. First consumer: source-uptick (airbytehq/airbyte#86356), which was blocked in review on this behaviour.

Problem

When an HttpResponseFilter maps a status (typically 401) to action: REFRESH_TOKEN_THEN_RETRY, HttpClient._handle_error_resolution calls authenticator.refresh_and_set_access_token() inside a try/except Exception that only logs a warning, then raises the normal backoff exception so the request is retried. Every retry lands in the same branch and triggers another refresh. When the refresh itself is rejected (OAuth password/refresh grant returns 400 invalid_grant / 401 invalid_client because the credential was rotated), the authenticator's own AirbyteTracedException(failure_type=config_error) was swallowed, so a single request cost max_retries + 1 failed logins (6 by default), and a sync with ~55 streams cost ~330 failed logins against the provider — enough to trigger account lockout on providers that lock after repeated failed logins. Before this action existed the same event cost exactly one failed login (fail-fast config_error).

Changes

http_client.py

• One token refresh per request: HttpClient tracks _token_refresh_outcomes: Dict[PreparedRequest, bool] (False = refresh attempted, True = refresh succeeded; recorded only when a refresh is actually attempted, cleared in _evict_key alongside _request_attempt_count).
• If the refresh raises AirbyteTracedException with FailureType.config_error (the authenticator's refresh_token_error_* classification), it is re-raised immediately — no backoff, no retry.
• A token-endpoint 4xx that does not match refresh_token_error_* (e.g. 401 invalid_client with the default refresh_token_error_status_codes=(400,)) is also a credential rejection: HttpClient._is_token_endpoint_rejection (4xx except 429) fails it fast as config_error ("OAuth token refresh request is rejected by the token endpoint."), 1 token request, 1 stream request. 429/5xx (DefaultBackoffException) and network errors stay transient.
• If the same request is rejected again after a refresh attempt, it fails instead of refreshing again, with a fixed user-facing message: config_error "Refreshed OAuth access token is rejected by the API." when the refresh had succeeded (the new token is rejected too), transient_error "API rejects the current OAuth access token and the token refresh failed." when the refresh itself failed transiently, so the platform can retry the sync instead of asking the user to fix their config. The filter's error_message (often a progress note such as "Token expired, refreshing...") is kept in internal_message, not shown as the terminal error.
• Unclassified refresh exceptions keep today's warn-and-retry-with-existing-token behaviour; the authenticator already backs off 429/5xx from the token endpoint internally for up to 300s. That path did not cause the failed-login loop.
• Authenticators without refresh_and_set_access_token keep the existing unbounded "normal retry" path (no state is recorded for them).

abstract_oauth.py / oauth.py

• refresh_and_set_access_token (base and SingleUseRefreshTokenOauth2Authenticator) now takes the existing class-level _token_refresh_lock with a double-check: if another thread replaced the access token while this one waited, the refresh is skipped and the request is retried with that token. Previously this forced-refresh path was the only refresh that bypassed the lock, so with the one-refresh bound the loser of a concurrent refresh race (single-use refresh tokens) would have failed hard instead of self-healing. The lock becomes an RLock because get_access_token already holds it while calling refresh_and_set_access_token.
• _current_access_token_or_none helper: DeclarativeOauth2Authenticator.access_token raises when no token has been set yet.
• This also fixes a live bug on main that the forced-refresh path had for SingleUseRefreshTokenOauth2Authenticator: two concurrent forced refreshes both POSTed the same single-use refresh token and emitted two config control messages, the second built on an already-spent token (affects connectors pairing refresh_token_updater with this action, e.g. greenhouse, reddit-ads). With the lock + skip only one refresh and one control message are emitted.
• HttpClient._auth_header_changed_since(request): before forcing a refresh, the Authorization header the rejected request was sent with is compared to the authenticator's current one; if it already differs (another thread/instance refreshed after this request went out), the refresh is skipped and the request is retried with the current token. This covers the case the in-authenticator snapshot cannot: a request that was already in flight with the old token when the refresh happened.

No new manifest field; no public interface changes.

Error-message provenance: a refresh rejected via refresh_token_error_* re-raises the authenticator's own AirbyteTracedException (names the provider error, e.g. invalid_grant); an unmatched 4xx and a second rejection after a refresh use the fixed messages above, with request/provider detail and the filter text in internal_message. The refresh-skip in refresh_and_set_access_token compares the token held by the same authenticator instance; for SingleUseRefreshTokenOauth2Authenticator the token lives in the connector config shared by every stream's instance, so it covers separate instances too. For per-instance token authenticators the HttpClient header check above is what catches an in-flight request whose token was replaced.

Behaviour matrix (per request, default max_retries=5)

Scenario	Before	After
401 → refresh OK → 200	1 token req, 2 stream reqs, records	unchanged
401 → refresh rejected (invalid_grant, matches refresh_token_error_*)	6 token reqs, 6 stream reqs, backoff sleeps, then config_error ("Exhausted available request attempts…"; declarative 401/403 filters map to config_error for non-FAIL actions)	1 token req, 1 stream req, no sleep, config_error (authenticator message)
401 → refresh rejected with unmatched 4xx (invalid_client)	same as above	1 token req, 1 stream req, no sleep, config_error (fixed message)
401 → refresh OK → 401	6 token reqs, 6 stream reqs	1 token req, 2 stream reqs, config_error, fixed message; filter text in internal_message
401 → refresh transient failure → 401	6 refresh attempts, 6 stream reqs	1 refresh attempt, 2 stream reqs, transient_error, fixed message
FAIL / RETRY / RATE_LIMITED actions	—	unchanged

Tests

unit_tests/sources/streams/http/test_http_client.py, using a real Oauth2Authenticator (with refresh_token_error_* configured) and requests_mock:

• refresh once then succeed: 1 token request, 2 stream requests, retry carries the new bearer token

• refresh rejected: 1 token request, 1 stream request, config_error, time.sleep never called

• second 401 after successful refresh: 1 token request, 2 stream requests, config_error with the filter's message

• transient refresh failure then 401 again: 1 refresh attempt, 2 stream requests, transient_error

• per-request state is evicted after success (_token_refresh_outcomes == {}) and a later request refreshes again

• existing REFRESH_TOKEN_THEN_RETRY / FAIL / RETRY tests unchanged and passing

• forced refresh is skipped when the rejected request's Authorization header no longer matches the authenticator's (refresh spy never called, retry carries the new token)

• refresh rejected outside the configured refresh_token_error_* (401 invalid_client): 1 token request, 1 stream request, config_error, fixed message

• token endpoint 5xx (DefaultBackoffException): stays transient_error, 2 stream requests

• transient refresh failure mocked at the requests boundary (ConnectionError, time.sleep patched) so the authenticator's own backoff retries are visible

• DeclarativeOauth2Authenticator end-to-end 401 → refresh → 200

unit_tests/sources/streams/http/requests_native_auth/test_requests_native_auth.py:

• refresh_and_set_access_token skips the refresh when another thread replaced the token while it waited on the lock; refreshes normally otherwise
• get_access_token → refresh_and_set_access_token re-entrancy on a worker thread with join(timeout=5) (a revert to a plain Lock fails instead of hanging)
• two SingleUseRefreshTokenOauth2Authenticator instances over one shared config: exactly 1 token request and 1 control message; the lock is held during refresh_access_token

Notes / adjacent (not done here)

• A refresh rejection that is not covered by the authenticator's refresh_token_error_* configuration surfaces as a raw requests.HTTPError and is still treated as an unclassified failure (warn, retry once with the old token, then fail). Classifying arbitrary 4xx token-endpoint responses as credential errors is left as a possible follow-up.
• source-uptick's test_refresh_token_error_handling (feat(source-uptick): add error classification, api_budget, concurrency, and certification metadata airbyte#86356) encodes the old repeat-refresh behaviour and needs updating in that PR to expect a single refresh and the authenticator's config_error.
• declarative_component_schema.yaml has no per-action prose for REFRESH_TOKEN_THEN_RETRY to document this in; only the generic action description exists.

Link to Devin session: https://app.devin.ai/sessions/7513e9b249074ff8b9b87cb398d91786
Open in Devin Desktop: https://app.devin.ai/desktop/session/7513e9b249074ff8b9b87cb398d91786?variant=devin

[devin-ai-integration]

I'll fix CI failures and address comments from users with write access. I'll skip comments containing "(aside)".

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1790165041-bounded-refresh-retry#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1790165041-bounded-refresh-retry

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Full)

4 691 tests  +14   4 679 ✅ +14   14m 2s ⏱️ +8s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit ca06fbd. ± Comparison against base commit d8d8e6f.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Fast)

4 688 tests  +14   4 676 ✅ +14   10m 1s ⏱️ +9s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit ca06fbd. ± Comparison against base commit d8d8e6f.

♻️ This comment has been updated with latest results.

[Anatolii Yatsuk (tolik0)]

The diagnosis is right, and the fail-fast re-raise at http_client.py:619-621 is what actually removes the lockout amplification. The main comment is on :586: the second-occurrence branch does more than the lockout fix requires, and a smaller version — skip the refresh and fall through to the normal retry handling rather than raising — keeps the whole benefit while leaving retry semantics, the failure type and the error message unchanged. It also removes the two defects that only exist because of the early raise, the most serious being that a revoked credential outside the connector's refresh_token_error_* configuration is currently reported as transient_error and retried indefinitely.

One correction to the description. The behaviour matrix gives the "Before" outcome for 401 → refresh rejected as transient_error/system_error. For every declarative 401 filter the prior outcome was config_error: HttpResponseFilter honours a declared failure_type only when the action is FAIL (http_response_filter.py:98), so a non-FAIL action falls through to the default mapping, and both 401 and 403 map to config_error (default_error_mapping.py:36-44). The old exhaustion path carried that through via failure_type=e.failure_type or FailureType.system_error (:361). All 18 filter sites across the five connectors declaring this action sit on 401 or 403, so none changes failure type on that path — apple-search-ads' explicitly declared transient_error was already being ignored. Confirmed by reading the same manifest through ConcurrentDeclarativeSource on a base install and a head install: config_error on both.

The PR fixes a second, live bug that it never claims. On the base commit, two concurrent forced refreshes on a single-use authenticator both POST the same refresh token and emit two control messages, the second built on an already-spent token — which a real single-use provider rejects with invalid_grant. Reproduced by racing two instances over one shared config: base emits two control messages, head emits one. That affects greenhouse and reddit-ads and is worth stating in the description.

On the scope: no changed file is under sources/declarative/, and the locking half reaches every OAuth connector through get_access_token, not just users of this action. Repo precedent on these files is fix(oauth) (#883, #1138) and fix(http) (#1126). fix(oauth): would describe it better than low-code.

On validation: the Test Connectors matrix is source-hardcoded-records, source-shopify, source-google-drive, destination-motherduck, source-intercom and source-pokeapi — none references OAuthAuthenticator, Oauth2Authenticator, refresh_token_updater or this action in non-test code, so green says nothing here. Suggest /prerelease pinned on source-apple-search-ads: largest exposure at 8 filters, four handlers at max_retries: 10, and grant_type: client_credentials (manifest.yaml:781) means repeated refreshes are idempotent and cannot strand a stored secret. Second choice source-hubspot. Not greenhouse or reddit-ads — both use single-use rotating refresh tokens and a credentialed run consumes the stored secret.

Check: destination-motherduck is unrelated: it fails identically on all 12 most recent PRs (#1163 through #1175) on a MotherDuck token problem.

[devin-ai-integration]

👍 On it. Thanks — taking the description correction (prior outcome on 401 → refresh rejected was config_error, not transient_error/system_error), the note about the concurrent single-use double-refresh fix, and the fix(oauth): title. The fall-through vs fail-fast choice is escalated to the DRI since fail-fast was the stated requirement; the inline threads track the rest. Prerelease validation on source-apple-search-ads noted for the DRI as well.

---

Devin session