feat: add REDUCE_PAGE_SIZE response action for dynamic page-size reduction

[Anatolii Yatsuk (tolik0)]

What

Adds a REDUCE_PAGE_SIZE response action to the declarative framework. When a response filter resolves to it, the retriever shrinks the page size and re-fetches the same page — same next-page token, same stream slice — instead of failing the sync.

Three commits: the action itself, then two fixes that came out of actually adopting it in a connector (see Adoption below).

Why

Several declarative connectors already expose a page-size config field whose only purpose is letting a user recover from an API that rejects large pages. Recovery today means: sync fails, user reads the error, lowers the setting, restarts the sync.

Four known adopters. All use CursorPagination with a working page_size_option, so each adoption is a response-filter action swap plus a page_size_reduction block:

• source-github (airbytehq/airbyte-internal-issues#16519) — 6 GraphQL streams get 502/504 when the query is too expensive. GitHubGraphQLErrorHandler mutates stream.page_size and returns RETRY, but HttpClient replays the same PreparedRequest, so the oversized first is already serialized into the body and the retry never benefits. Only a later page picks up the smaller size, and any non-502/504 response resets it. This blocked the manifest-only migration.
• source-amazon-seller-partner ListFinancialEvents — fails on 400/InvalidInput ("the response exceeds the maximum number of transactions or 10 MB") with an error telling the user to "try reducing it to a smaller value (e.g., 50, 25, 10, or even 1 for very high-volume accounts)". A manual binary search across sync restarts.
• source-zendesk-support ticket_comments — retries 504 with exponential backoff at an unchanged per_page, so the retries cannot succeed. Its own spec says the fix is "lower values may help prevent timeouts on large datasets".
• source-facebook-pages — fails on "Please reduce the amount of data you're asking for", and both the spec field and the public docs instruct the user to decrease page_size by hand.

How

The signal path mirrors RESET_PAGINATION:

HttpResponseFilter action: REDUCE_PAGE_SIZE → HttpClient._handle_error_resolution raises PageSizeReductionRequiredException → the exception is not in TRANSIENT_EXCEPTIONS, so it escapes the backoff decorators untouched → SimpleRetriever._read_pages catches it, reduces, and continues without advancing the token. CompositeErrorHandler short-circuits on the new action so it is not swallowed when nested.

retriever:
  type: SimpleRetriever
  page_size_reduction:
    type: PageSizeReduction
    reduction_factor: 2       # default
    minimum_page_size: 1      # default
    max_attempts: 5           # default
    reset_policy: NEVER       # default; or AFTER_SUCCESSFUL_PAGE
  requester:
    error_handler:
      type: DefaultErrorHandler
      response_filters:
        - http_codes: [502, 504]
          action: REDUCE_PAGE_SIZE
          failure_type: transient_error

reset_policy defaults to NEVER because restoring the configured size after every good page re-triggers the error on each subsequent page and roughly doubles request volume.

Thread safety

model_to_component_factory builds one retriever per stream and hands that single object to DeclarativePartitionFactory, which reuses it for every partition; partitions are read concurrently by a thread pool. So the reduction state lives in a PageSizeReducer constructed inside _read_pages, like PaginationTracker — no mutable page-size state on the retriever, paginator, or pagination strategy. The effective size travels as an optional page_size_override keyword passed only when set, so out-of-tree Paginator / PaginationStrategy subclasses that do not accept it keep working. There is a test with two partitions on threads asserting the healthy one keeps its configured size while the failing one runs reduced.

Correctness fix in OffsetIncrement

Its stop condition compared the returned record count against the configured page size. Reduced to 50 with a configured 100, a full 50-record page would hit 50 < 100, end pagination, and silently drop the rest of the partition. The stop condition now honors the override; the offset math already advanced by the actual last_page_size, so that part was fine.

Rejected at config time, not mid-sync

PageIncrement (the token is a page number, so halving moves record boundaries and skips records — and its page size is also its stop condition), paginators other than DefaultPaginator, a missing page_size_option, query-properties chunking (earlier chunks are already emitted, so a retry would duplicate), LazySimpleRetriever, and declaring the action without a page_size_reduction block. Each raises with an actionable message naming the stream.

Termination

Every reduction strictly decreases the size or raises. Both the minimum_page_size floor and max_attempts raise AirbyteTracedException with FailureType.transient_error. The attempt counter is never reset, so the worst case is pages + max_attempts requests per partition.

Adoption, and the two things it caught

airbytehq/airbyte#85822 migrates all six source-github GraphQL streams onto this action. Building it surfaced two defects in the first commit here, both fixed rather than worked around in the connector:

1. c5fafe32 — the strategy allowlist rejected every CustomPaginationStrategy. The check was by type, which excluded exactly the streams the feature was built for: source-github's nested GraphQL traversal cannot be expressed by a built-in strategy. What actually matters is not whether the CDK recognizes the strategy but whether the strategy can be told the reduced size — and a custom strategy is written by the same person enabling the reduction. A custom strategy is now accepted when its next_page_token takes a page_size_override keyword (or **kwargs), checked by signature at config time so one that would raise TypeError on the first reduction is rejected up front with an actionable message.
2. fed78ad4 — a non-integer page size crashed the reducer. reduce() did current_page_size // reduction_factor on whatever get_page_size returned. Built-in strategies always return an int; a custom one can return anything, and a string page size — which is what a custom component gets when a manifest field is left uninterpolated — failed mid-sync with TypeError: unsupported operand type(s) for //: 'str' and 'float'. Now a config error naming the type.

Scope

Page size only. Request-window reduction (airbytehq/airbyte-internal-issues#17173) is a separate axis — it re-slices the datetime range and touches cursor state. This action never changes slice boundaries. The two are intended to compose, and this being the second occupant of the ResponseAction → signal-exception → _read_pages seam should make that one a copy of an established pattern rather than a third mechanism.

Relationship to #1056

Supersedes the draft in #1056, which will be closed. Same action name and same exception, different mechanics. #1056 mutated _page_size on the shared strategy instance, so one partition's success wiped another's reduction mid-flight; its reduce loop was unbounded, so a permanently failing endpoint looped forever at page size 1; its OffsetIncrement change routed the reduced size through get_page_size() only, leaving the stop-condition data loss above in place; it let PageIncrement reduce; it did not update the CompositeErrorHandler short-circuit list; and its isinstance(self._paginator, DefaultPaginator) guard silently no-oped under PaginatorTestReadDecorator, so Connector Builder test reads skipped reduction entirely.

Testing

• unit_tests/sources/declarative/retrievers/test_page_size_reducer.py (15 tests) — factor/floor/attempt arithmetic, reset policies, and the non-integer page size guard.
• test_simple_retriever.py — the retry re-fetches the same page, the retry request actually carries the reduced size (asserted on the outgoing request, not on strategy state), reset-policy behavior, floor and max_attempts exhaustion, the action raising when no page_size_reduction is configured, and the two-thread partition-isolation test.
• Paginator and strategy tests for override forwarding, including PaginatorTestReadDecorator.
• test_offset_increment.py pins the stop-condition fix.
• test_model_to_component_factory.py — 10 tests for the config-time rejections, including a custom strategy that accepts the override and one that does not.
• test_concurrent_declarative_source.py — manifest-level HttpMocker read asserting request 1 is first=100, the 502 is followed by request 2 to the same cursor with first=50, and all records arrive.

CI on c5fafe32: 4430 tests, 0 failures. Locally mypy, ruff check and ruff format --check are clean, and the full suite is green except two pre-existing test_file_based_scenarios.py unstructured failures confirmed identical on a clean baseline.

The Check: destination-motherduck connector check fails, and is unrelated: its unit tests pass and only the FAST standard tests fail, this change touches declarative source pagination only, and the same check fails on unrelated branches (devin/1787781944-record-expander-truncated-list, 6 consecutive runs).

Draft, because

poe assemble could not run locally (Docker daemon down), so airbyte_cdk/sources/declarative/models/declarative_component_schema.py was written by hand in codegen style. It loads and behaves correctly, but needs poe assemble to confirm a zero diff — the enum class name generated for reset_policy in particular.

Known gap, not addressed here

When reduction bottoms out, the terminal message is "The source is still failing with the smallest page the connector is allowed to request (1 records per page)… Try syncing fewer streams at once, or contact the API provider." It does not name the stream and suggests an unrelated remedy. The connector-specific message it replaces named the stream and pointed at the page-size setting. PageSizeReduction should accept a connector-supplied error_message; the same complaint applies to PaginationTracker's terminal error and was raised in the airbytehq/airbyte-internal-issues#17173 triage. Happy to fold it into this PR if reviewers prefer.

🤖 Generated with Claude Code

[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@tolik0/cdk/reduce-page-size#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 tolik0/cdk/reduce-page-size

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 (Fast)

4 428 tests  +65   4 416 ✅ +65   9m 17s ⏱️ +8s
    1 suites ± 0      12 💤 ± 0 
    1 files   ± 0       0 ❌ ± 0 

Results for commit fed78ad. ± Comparison against base commit 83933f1.

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

4 431 tests   4 419 ✅  9m 49s ⏱️
    1 suites     12 💤
    1 files        0 ❌

Results for commit fed78ad.

♻️ This comment has been updated with latest results.

[Anatolii Yatsuk (tolik0)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/34491835004

[Anatolii Yatsuk (tolik0)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/34504922288

[Anatolii Yatsuk (tolik0)]

/prerelease

Prerelease Job Info

This job triggers the publish workflow with default arguments to create a prerelease.

Prerelease job started... Check job output.

✅ Prerelease workflow triggered successfully.

View the publish workflow run: https://github.com/airbytehq/airbyte-python-cdk/actions/runs/34507001618