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

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -2658,6 +2658,7 @@ definitions:
           - RESET_PAGINATION
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
+          - REDUCE_PAGE_SIZE
         examples:
           - SUCCESS
           - FAIL
@@ -2666,6 +2667,7 @@ definitions:
           - RESET_PAGINATION
           - RATE_LIMITED
           - REFRESH_TOKEN_THEN_RETRY
+          - REDUCE_PAGE_SIZE
       failure_type:
         title: Failure Type
         description: Failure type of traced exception if a response matches the filter.
@@ -4223,6 +4225,9 @@ definitions:
       pagination_reset:
         description: Describes what triggers pagination reset and how to handle it.
         "$ref": "#/definitions/PaginationReset"
+      page_size_reduction:
+        description: Describes how the page size is reduced when an error handler resolves to the REDUCE_PAGE_SIZE action.
+        "$ref": "#/definitions/PageSizeReduction"
       ignore_stream_slicer_parameters_on_paginated_requests:
         description: If true, the partition router and incremental request options will be ignored when paginating requests. Request options set directly on the requester will not be ignored.
         type: boolean
@@ -4278,6 +4283,57 @@ definitions:
         enum: [PaginationResetLimits]
       number_of_records:
         type: integer
+  PageSizeReduction:
+    title: Page Size Reduction
+    description: >-
+      Describes how the page size is reduced when an error handler resolves to the REDUCE_PAGE_SIZE action. On such
+      a response, the connector re-issues the same page with a smaller page size instead of retrying the identical
+      request. Only supported on a DefaultPaginator that has a page_size_option and whose pagination_strategy is
+      CursorPagination, OffsetIncrement, or a CustomPaginationStrategy whose next_page_token accepts a
+      page_size_override keyword argument. PageIncrement is not supported because a smaller page size moves every
+      following page boundary and would skip records.
+    type: object
+    required:
+      - type
+    properties:
+      type:
+        type: string
+        enum: [PageSizeReduction]
+      reduction_factor:
+        title: Reduction Factor
+        description: Divisor applied to the page size on each reduction. The new page size is floor(current page size / reduction factor).
+        type: number
+        default: 2
+        exclusiveMinimum: 1
+        examples:
+          - 2
+          - 4
+      minimum_page_size:
+        title: Minimum Page Size
+        description: Page size below which the connector stops reducing and fails the sync.
+        type: integer
+        default: 1
+        minimum: 1
+        examples:
+          - 1
+          - 10
+      max_attempts:
+        title: Maximum Reduction Attempts
+        description: Maximum number of page size reductions allowed while reading a single partition. Exceeding it fails the sync with a transient error.
+        type: integer
+        default: 5
+        minimum: 1
+      reset_policy:
+        title: Reset Policy
+        description: >-
+          When to restore the page size configured on the pagination strategy. NEVER keeps the reduced page size for
+          the rest of the partition. AFTER_SUCCESSFUL_PAGE restores it as soon as one page succeeds, which can mean
+          hitting the same error again on every page.
+        type: string
+        enum:
+          - NEVER
+          - AFTER_SUCCESSFUL_PAGE
+        default: NEVER
   GzipDecoder:
     title: gzip
     description: Select 'gzip' for response data that is compressed with gzip. Requires specifying an inner data type/decoder to parse the decompressed data.

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -714,6 +714,7 @@ class Action(Enum):
     RESET_PAGINATION = "RESET_PAGINATION"
     RATE_LIMITED = "RATE_LIMITED"
     REFRESH_TOKEN_THEN_RETRY = "REFRESH_TOKEN_THEN_RETRY"
+    REDUCE_PAGE_SIZE = "REDUCE_PAGE_SIZE"
 
 
 class FailureType(Enum):
@@ -735,6 +736,7 @@ class HttpResponseFilter(BaseModel):
             "RESET_PAGINATION",
             "RATE_LIMITED",
             "REFRESH_TOKEN_THEN_RETRY",
+            "REDUCE_PAGE_SIZE",
         ],
         title="Action",
     )
@@ -1410,6 +1412,37 @@ class PaginationResetLimits(BaseModel):
     number_of_records: Optional[int] = None
 
 
+class ResetPolicy(Enum):
+    NEVER = "NEVER"
+    AFTER_SUCCESSFUL_PAGE = "AFTER_SUCCESSFUL_PAGE"
+
+
+class PageSizeReduction(BaseModel):
+    type: Literal["PageSizeReduction"]
+    reduction_factor: Optional[float] = Field(
+        2,
+        description="Divisor applied to the page size on each reduction. The new page size is floor(current page size / reduction factor).",
+        examples=[2, 4],
+        title="Reduction Factor",
+    )
+    minimum_page_size: Optional[int] = Field(
+        1,
+        description="Page size below which the connector stops reducing and fails the sync.",
+        examples=[1, 10],
+        title="Minimum Page Size",
+    )
+    max_attempts: Optional[int] = Field(
+        5,
+        description="Maximum number of page size reductions allowed while reading a single partition. Exceeding it fails the sync with a transient error.",
+        title="Maximum Reduction Attempts",
+    )
+    reset_policy: Optional[ResetPolicy] = Field(
+        "NEVER",
+        description="When to restore the page size configured on the pagination strategy. NEVER keeps the reduced page size for the rest of the partition. AFTER_SUCCESSFUL_PAGE restores it as soon as one page succeeds, which can mean hitting the same error again on every page.",
+        title="Reset Policy",
+    )
+
+
 class CsvDecoder(BaseModel):
     type: Literal["CsvDecoder"]
     encoding: Optional[str] = "utf-8"
@@ -3222,6 +3255,10 @@ class SimpleRetriever(BaseModel):
         None,
         description="Describes what triggers pagination reset and how to handle it.",
     )
+    page_size_reduction: Optional[PageSizeReduction] = Field(
+        None,
+        description="Describes how the page size is reduced when an error handler resolves to the REDUCE_PAGE_SIZE action.",
+    )
     ignore_stream_slicer_parameters_on_paginated_requests: Optional[bool] = Field(
         False,
         description="If true, the partition router and incremental request options will be ignored when paginating requests. Request options set directly on the requester will not be ignored.",

airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py

@@ -138,6 +138,9 @@
     DEPRECATION_LOGS_TAG,
     BaseModelWithDeprecations,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    Action as HttpResponseFilterActionModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     Action1 as PaginationResetActionModel,
 )
@@ -384,6 +387,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     PageIncrement as PageIncrementModel,
 )
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    PageSizeReduction as PageSizeReductionModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     PaginationReset as PaginationResetModel,
 )
@@ -579,6 +585,10 @@
     LocalFileSystemFileWriter,
     NoopFileWriter,
 )
+from airbyte_cdk.sources.declarative.retrievers.page_size_reducer import (
+    PageSizeReduction,
+    PageSizeResetPolicy,
+)
 from airbyte_cdk.sources.declarative.retrievers.pagination_tracker import PaginationTracker
 from airbyte_cdk.sources.declarative.schema import (
     ComplexFieldType,
@@ -3639,6 +3649,12 @@ def _get_url(req: Requester) -> str:
                     f"LazySimpleRetriever only supports JsonDecoder. Found: {model.decoder.type}."
                 )
 
+            if model.page_size_reduction:
+                raise ValueError(
+                    f"`page_size_reduction` is not supported when reading a parent stream lazily. Remove either "
+                    f"`page_size_reduction` or the parent stream's `lazy_read_pointer` for stream {name}."
+                )
+
             return LazySimpleRetriever(
                 name=name,
                 paginator=paginator,
@@ -3675,10 +3691,115 @@ def _get_url(req: Requester) -> str:
             pagination_tracker_factory=self._create_pagination_tracker_factory(
                 model.pagination_reset, cursor
             ),
+            page_size_reduction=self._create_page_size_reduction(model, name, query_properties),
             post_pagination_filter=post_pagination_filter,
             parameters=model.parameters or {},
         )
 
+    def _create_page_size_reduction(
+        self,
+        model: SimpleRetrieverModel,
+        name: str,
+        query_properties: Optional[QueryProperties],
+    ) -> Optional[PageSizeReduction]:
+        # A CustomRequester does not necessarily define an error handler
+        error_handler = getattr(model.requester, "error_handler", None)
+        if self._uses_reduce_page_size_action(error_handler) and not model.page_size_reduction:
+            raise ValueError(
+                f"Stream {name} has a response filter with the REDUCE_PAGE_SIZE action but the retriever does not "
+                f"define `page_size_reduction`. Add a `page_size_reduction` block to the retriever."
+            )
+
+        if not model.page_size_reduction:
+            return None
+
+        self._validate_page_size_reduction_is_supported(model, name, query_properties)
+
+        reset_policy = model.page_size_reduction.reset_policy
+        return PageSizeReduction(
+            reduction_factor=model.page_size_reduction.reduction_factor,  # type: ignore[arg-type]  # the schema defines a default
+            minimum_page_size=model.page_size_reduction.minimum_page_size,  # type: ignore[arg-type]  # the schema defines a default
+            max_attempts=model.page_size_reduction.max_attempts,  # type: ignore[arg-type]  # the schema defines a default
+            reset_policy=PageSizeResetPolicy(getattr(reset_policy, "value", reset_policy)),
+        )
+
+    def _validate_page_size_reduction_is_supported(
+        self,
+        model: SimpleRetrieverModel,
+        name: str,
+        query_properties: Optional[QueryProperties],
+    ) -> None:
+        """
+        Page size reduction re-issues the same page with a smaller page size. That is only correct when the next
+        page does not depend on the page size, and it only has an effect when the paginator injects the page size
+        in the request. A custom pagination strategy is accepted when it can receive the reduced page size, which
+        is checked by inspecting its signature rather than by recognizing its type.
+        """
+        if query_properties:
+            raise ValueError(
+                f"`page_size_reduction` cannot be used together with query properties on stream {name}. Records "
+                f"from the earlier property chunks have already been emitted when a chunk asks for a smaller page, "
+                f"so retrying the page would emit them twice."
+            )
+
+        if not isinstance(model.paginator, DefaultPaginatorModel):
+            raise ValueError(
+                f"`page_size_reduction` requires a DefaultPaginator on stream {name} so that the connector can "
+                f"send a smaller page size."
+            )
+
+        if not model.paginator.page_size_option:
+            raise ValueError(
+                f"`page_size_reduction` requires `page_size_option` on the paginator of stream {name}: without it "
+                f"the connector cannot tell the API to send a smaller page."
+            )
+
+        strategy = model.paginator.pagination_strategy
+        if isinstance(strategy, PageIncrementModel):
+            raise ValueError(
+                f"`page_size_reduction` does not support the PageIncrement pagination strategy used by stream "
+                f"{name}. Pages are addressed as page number * page size, so a smaller page size shifts every "
+                f"following page boundary and would skip records. Use OffsetIncrement or CursorPagination."
+            )
+        if isinstance(strategy, CustomPaginationStrategyModel):
+            # A custom strategy is written by the same person enabling the reduction, so the
+            # question is not whether we recognize it but whether it can be told the reduced
+            # page size. Checking the signature keeps a strategy that would raise TypeError
+            # mid-sync from being accepted at config time.
+            custom_class = self._get_class_from_fully_qualified_class_name(strategy.class_name)
+            parameters = inspect.signature(custom_class.next_page_token).parameters
+            accepts_override = "page_size_override" in parameters or any(
+                parameter.kind == inspect.Parameter.VAR_KEYWORD for parameter in parameters.values()
+            )
+            if not accepts_override:
+                raise ValueError(
+                    f"`page_size_reduction` requires the custom pagination strategy "
+                    f"{strategy.class_name} used by stream {name} to accept a `page_size_override` keyword "
+                    f"argument in `next_page_token`, so that it can honor the reduced page size. Add "
+                    f"`page_size_override: Optional[int] = None` to its signature; a strategy that does not "
+                    f"use its page size as a stop condition can ignore the value."
+                )
+        elif not isinstance(strategy, (CursorPaginationModel, OffsetIncrementModel)):
+            raise ValueError(
+                f"`page_size_reduction` only supports the CursorPagination, OffsetIncrement and "
+                f"CustomPaginationStrategy pagination strategies. Stream {name} uses "
+                f"{type(strategy).__name__}."
+            )
+
+    def _uses_reduce_page_size_action(self, error_handler: Any) -> bool:
+        if isinstance(error_handler, CompositeErrorHandlerModel):
+            return any(
+                self._uses_reduce_page_size_action(nested)
+                for nested in error_handler.error_handlers
+            )
+        if isinstance(error_handler, DefaultErrorHandlerModel):
+            return any(
+                response_filter.action == HttpResponseFilterActionModel.REDUCE_PAGE_SIZE
+                for response_filter in error_handler.response_filters or []
+            )
+        # A CustomErrorHandler can return any action and we cannot inspect it, so we do not validate it.
+        return False
+
     def _create_pagination_tracker_factory(
         self, model: Optional[PaginationResetModel], cursor: Cursor
     ) -> Callable[[], PaginationTracker]:

airbyte_cdk/sources/declarative/requesters/error_handlers/composite_error_handler.py

@@ -71,6 +71,7 @@ def interpret_response(
                 ResponseAction.RETRY,
                 ResponseAction.IGNORE,
                 ResponseAction.RESET_PAGINATION,
+                ResponseAction.REDUCE_PAGE_SIZE,
             ]:
                 return matched_error_resolution