fix(source-salesforce): let REST streams refresh their token and restart the query when the session is invalidated mid-pagination - #84302
Conversation
…dated mid-pagination A Salesforce query locator belongs to the session that created it. When that session is invalidated while a REST stream is paginating, the error handler refreshes the token and retries the same `nextRecordsUrl`, which can never succeed: the stream exhausts its retries and fails the sync. This is routine with Refresh Token Rotation, where every token exchange ends the previous session. The connector refreshes proactively every 30 minutes, so any REST stream whose read takes longer than that is affected. Streams on the Bulk API path are not. The error handler now raises for a locator request so the caller can tell the difference between an ordinary expired session and a dead locator, and `_read_pages` restarts the query after the last primary key it read for that property chunk. A restart that reads nothing before its locator expires again fails the attempt rather than restarting from the same position forever, and the paths that cannot resume raise a transient `AirbyteTracedException` instead of a bare error. The resume position is only sound for a query ordered by primary key, so `IncrementalRestSalesforceStream` now emits `ORDER BY <pk> ASC`. Note this is a new clause for those streams, including when they are read in full refresh mode via the no-cursor branch; only streams without a replication key, which use `RestSalesforceStream.request_params`, already ordered this way.
👋 Welcome to Airbyte!Thank you for your contribution from jtruty/airbyte! We're excited to have you in the Airbyte community. If you have any questions, feel free to ask in the PR comments or join our Slack community. 💡 Show Tips and TricksPR Slash CommandsAs needed or by request, Airbyte Maintainers can execute the following slash commands on your PR:
Tips for Working with CI
📚 Show Repo GuidanceHelpful Resources
|
REST streams authenticated with `TokenAuthenticator(sf_object.access_token)`, a token string captured when the stream was built, and their error handler was built without a token provider. They could therefore neither trigger a proactive refresh nor observe one, and `INVALID_SESSION_ID` could not force a refresh, so a REST read that outlived the org session timeout had no way to recover. They now read from the same `SalesforceTokenProvider` the Bulk path uses, and the error handler is given that provider so the existing force-refresh and fail-fast paths apply. This must land together with the query restart in the same PR. On its own it makes REST streams start refreshing at the 30 minute mark, which rotates the token and invalidates any locator they hold, so long REST reads that survive today would begin failing without the restart to recover them.
James Truty (jtruty)
changed the title
… streams The custom authenticator added in the previous commit duplicated what `BearerAuthenticator` already does: it is an `AuthBase` and reads the token provider on every call, which is how the Bulk path is already wired.
|
↪️ Triggering Reason: Draft community fix; the only red checks are the no-credentials connector test on a fork PR and the progressive rollout gate, neither of which this diff affects. Prove-fix validation is the next pipeline step for: |
|
🧪 Fix Validation (
|
Airbyte Support Bot (airbyte-support-bot)
added
the
hyd-prove
|
|
↪️ Triggering Reason: |
Reviewing PR for connector safety and quality.
|
🛡️ AI PR Review Report🟡 Review Action: No Automated Review Submitted (UNKNOWN)Two enforced gates could not be resolved from the available evidence, so no automated approval or change request is submitted. A maintainer decision is needed. 🟠 Risk Level: 4 / 5Functional runtime change to REST pagination and auth: streams.py adds per-chunk query restart state and an unconditional ORDER BY <primary_key> ASC to REST SOQL queries, rate_limiting.py raises a new QueryLocatorExpiredException on INVALID_SESSION_ID for nextRecordsUrl requests, and source.py swaps the captured TokenAuthenticator for a provider-backed BearerAuthenticator. Unit coverage is good but the fixed path was never exercised against a live org and the query-plan impact of the new ORDER BY on large orgs is unmeasured. Gate Summary8 of 12 gates passed. Non-passing gates:
No gate failed outright, so there is nothing to remediate before merge from this review's perspective. To resolve the two UNKNOWN gates:
📋 PR Details
🔍 Gate Evaluation DetailsPR Hygiene — ✅ PASS (enforced) Semantic title, detailed body describing the failure and fix, version bumped in metadata.yaml and pyproject.toml (2.8.1 -> 2.8.2), docs changelog row added for 2.8.2, no unresolved human review comments (all existing comments are bots). Code Hygiene — ✅ PASS (warning-only) Connector code and unit tests changed together; new exception, restart-state fields and guards carry explanatory docstrings/comments; no debug leftovers, dead code or TODOs in the diff. Test Coverage — ✅ PASS (enforced) unit_tests/api_test.py adds 7 targeted tests: authenticator reads the current token, error handler can force a refresh, restart on locator/session expiry, restart stitched across property chunks, repeated restarts while making progress, transient error when restart is impossible, and give-up when a restart makes no progress. 145 unit tests pass in CI. Code Security — ✅ PASS (enforced) No secrets logged or added; access tokens are read through SalesforceTokenProvider instead of being captured once; no new network hosts and allowedHosts is unchanged. Note: the restart filter interpolates the last primary key into SOQL ( Per-Record Performance — The record loop in streams.py now evaluates the supports_query_restart property and tracks last_primary_key per record. Overhead is small but the property (isinstance check plus two set lookups) could be hoisted out of the loop. Breaking Dependencies — ✅ PASS (warning-only) No dependency or CDK version changes; pyproject.toml only bumps the connector version. Backwards Compatibility — ✅ PASS (enforced) No spec, schema, primary key, cursor field or state format change; no stream removed or renamed. Patch bump 2.8.2 is appropriate and no releases.breakingChanges entry or migration guide is required. Forwards Compatibility — Pagination control flow changed, but the restart state (last_primary_key, restart_after_primary_key, last_restart_primary_key) lives only in memory on PropertyChunk, so persisted state stays compatible and a rollback to 2.8.1 is safe. Restarting mid-slice can re-emit records already emitted in that slice, so dedup relies on primary-key dedup at the destination. Behavioral Changes — Three user-visible behavior changes: (1) REST SOQL queries now always append ORDER BY <primary_key> ASC for restart-eligible streams, whose query-plan impact on large orgs is untested (acknowledged in the PR body); (2) INVALID_SESSION_ID on a nextRecordsUrl now restarts the query, or raises a transient AirbyteTracedException when it cannot restart, instead of retrying a URL that can only fail; (3) REST streams authenticate through SalesforceTokenProvider, so a mid-sync refresh (including refresh token rotation) is now picked up. Out-of-Scope Changes — ✅ PASS (warning-only) All changes are confined to airbyte-integrations/connectors/source-salesforce/** and docs/integrations/sources/salesforce.md. CI Checks — ❓ UNKNOWN (enforced) Lint, Format Check, Build and Verify Artifacts, Pre-Release Checks, Check Changelog Updated and the 145 unit tests all pass. Live / E2E Tests — ❓ UNKNOWN (enforced)
📚 Evidence Consulted
❓ How to Respond
|
3 participants
What
A Salesforce query locator belongs to the session that created it. When that session is invalidated while a REST stream is paginating,
SalesforceErrorHandlerreturnsRETRY, so the CDK retries the samenextRecordsUrl(and on the REST path it could not even refresh the token first, see below). A new session cannot resume another session's locator, so every retry fails withINVALID_SESSION_ID, the stream exhausts its attempts, and the sync fails.This is routine with Refresh Token Rotation enabled, where each token exchange ends the previous session. Since 2.7.20 the connector re-logs in roughly every 30 minutes during a sync, so any stream on the REST path whose read outlives that interval will have a locator open when a refresh lands. Streams using the Bulk API are unaffected, so this shows up on objects that cannot use Bulk, for example anything containing a compound field such as an address.
Diagnosed on a production connection where every attempt died the same way: the read progresses normally, then
Refreshing Salesforce OAuth token (N s since last login)is followed 2-5 seconds later by repeatedSalesforce session expired or invalid. Token has been refreshed.until the stream gives up. 13 of 13 attempts on one job, always the same stream, while every Bulk-path stream in the same sync completed. Another connection with a larger and longer all-Bulk sync re-logged in twice mid-run in the same window with zero session errors, which isolates it to the REST pagination path rather than to size or duration.Pre-RTR the same re-login was harmless: two earlier jobs on the same connection re-logged in mid-sync with no rotation persisted, no session errors, and both succeeded. So the trigger is the rotation ending the session, not the refresh itself.
How
Two defects, fixed together because neither is safe or sufficient alone (see the ordering note below).
REST streams could not refresh. They authenticated with
TokenAuthenticator(sf_object.access_token)(source.py), a token captured when the stream was built, and theirHttpClienterror handler was built without atoken_provider(streams.py). So they never triggeredrefresh_access_token_if_stale(), never observed a refresh, andINVALID_SESSION_IDcould not force one. Only the Bulk path had a provider, wired throughBearerAuthenticator. They now use the CDK'sBearerAuthenticatorwith the sameSalesforceTokenProviderthe Bulk path uses, and the error handler is given that provider so the existing force-refresh and fail-fast paths apply.An orphaned locator was retried rather than restarted.
SalesforceErrorHandler.interpret_responseraisesQueryLocatorExpiredExceptionwhen the failing request is a locator request, after the existing token refresh. An ordinary expired session is still retried exactly as before. Detection matches on the URL path only so a SOQL query string cannot produce a false positive, following the existing_is_bulk_job_status_checkpattern._read_pagescatches it and restarts the query after the last primary key read for that property chunk, tracked per chunk so chunks resume independently.UNSUPPORTED_FILTERING_STREAMS, parent objects, or nothing read yet) raise a transientAirbyteTracedExceptioninstead of a bare error, preserving the readable failure the previous RETRY resolution produced.IncrementalRestSalesforceStream.request_paramsnow emitsORDER BY <pk> ASC, which is what makes the resume position sound.Review guide
source_salesforce/source.py— the provider-backedBearerAuthenticatoringenerate_streams.source_salesforce/streams.py— thetoken_providernow passed to the REST error handler, plus_read_pagesrestart and guard,supports_query_restart, and theORDER BYinIncrementalRestSalesforceStream.request_params.source_salesforce/rate_limiting.py— the new exception and the locator detection.unit_tests/api_test.py— five pagination tests plus two for the authenticator and error-handler wiring.Two things worth a second opinion:
The
ORDER BYis a new clause for most REST streams, not an alignment with existing behaviour. Streams without a replication key already ordered by primary key viaRestSalesforceStream.request_params. Streams with one did not, in either the sliced branch or the no-cursor branch used when they are read in full refresh mode. On a very large object this could lead the query optimizer to favour the Id index over the cursor index and surface asQUERY_TIMEOUT.GET /services/data/vXX.X/query/?explain=<SOQL>reportsleadingOperationTypeandrelativeCostper candidate plan and would settle it in two calls, but that needs a large org I do not have access to. It cannot be added lazily either: a resume position requires deterministic order from the query's first page.Alternative considered and rejected: re-read the current slice from its start and let primary-key dedup absorb the overlap. That needs no
ORDER BYand carries no plan risk, but it livelocks on exactly the case this fixes, since any slice whose read exceeds the refresh interval can never complete, and for a full refresh stream the slice is the whole stream. It is also unsafe here: the multi-chunk stitcher counts parts per primary key, so re-reading an already-counted chunk can emit records missing other chunks' fields. Happy to switch approach if you would rather not change query shape on this connector.Also noted while working here, left alone as out of scope: when a parent-object
WHEREclause is present inRestSalesforceStream.request_params,ORDER BYis concatenated without a separating space. It looks unreachable today because those streams are inUNSUPPORTED_FILTERING_STREAMS.What made this reachable, and why it is rare
The trigger is narrower than "a long REST read":
refresh_access_token_if_stale()was only reachable from the Bulk path.SalesforceTokenProvideris wired intoBulkSalesforceStream._instantiate_declarative_streamviaBearerAuthenticator(streams.py:647), andSalesforceTokenProvider.get_token()is its only caller.TokenAuthenticator(sf_object.access_token), a token captured once at construction, so they neither triggered a proactive refresh nor observed one.HttpClienterror handler had notoken_provider, so onINVALID_SESSION_IDtheforce_refresh()call was skipped — theToken has been refreshed.in that message was inaccurate on this path.So the failure needs a Bulk stream requesting a token past the refresh interval while a REST stream is mid-pagination. That matches what we see in production: a connection with three heavy concurrent Bulk streams failed on every attempt, while another making 300+ locator calls across a 90-minute sync, with little late Bulk activity, has never hit it in 30 days.
Why these two are one PR and not two
I originally split them and then merged them back, because the split creates a state that must never exist: REST streams that refresh but cannot restart.
refresh_access_token_if_stale(), so a long REST read that today never refreshes would start rotating at the 30 minute mark and invalidate its own open locator. Syncs that are healthy today would begin failing.INVALID_SESSION_IDwith no Bulk stream involved; the restarted query presents the same dead token, makes no progress, and the guard fails the attempt.Splitting would let that bad state arise from a wrong merge order or from reverting only the restart half later. Together they are order-independent and revert as a unit. If you would still rather review them separately, say so and I will split with the ordering constraint made explicit.
User Impact
REST-path streams whose reads outlive the token refresh interval can complete instead of failing. Before this change, with RTR enabled such a stream could not finish at all, and the job burned its full retry budget re-reading the same data.
Behaviour changes for users:
ORDER BY <pk> ASC. Emitted record order changes accordingly; state logic is order-insensitive, so there is no correctness impact.lookback_window.Can this PR be safely reverted and rolled back?