fix(low-code): report why the API rejected the last window or page

airbyte_cdk/sources/declarative/retrievers/page_size_reducer.py

@@ -110,10 +110,11 @@ def page_size_override(self) -> Optional[int]:
         """
         return self._current_page_size
 
-    def reduce(self) -> None:
+    def reduce(self, error_message: Optional[str] = None) -> None:
         """
         Shrink the page size used for the next request. Raises once the page size cannot be shrunk any further
-        so that an endpoint that keeps failing does not loop forever.
+        so that an endpoint that keeps failing does not loop forever. Those errors include `error_message`, the
+        error handler's message for the rejected page.
         """
         if self._configured_page_size is None:
             raise AirbyteTracedException(
@@ -139,14 +140,15 @@ def reduce(self) -> None:
                 failure_type=FailureType.config_error,
             )
 
+        rejection = f"; the API rejected the page: {error_message}" if error_message else ""
         reduced_page_size = max(
             self._config.minimum_page_size,
             int(current_page_size // self._config.reduction_factor),
         )
         if reduced_page_size >= current_page_size:
             # Nothing left to give up on the page size. Whether that is the end of the read is
             # `retries_at_minimum_page_size`'s call, not this branch's: the response may still be transient.
-            self._retry_at_minimum_page_size(current_page_size)
+            self._retry_at_minimum_page_size(current_page_size, rejection)
             return
 
         self._attempts += 1
@@ -157,7 +159,7 @@ def reduce(self) -> None:
             # succeeding restarts this counter on each of them, under either reset policy, and reads to the
             # end however many pages it has; a partition where nothing gets through burns the budget here.
             raise AirbyteTracedException(
-                internal_message=f"Stream {self._stream_name} reduced its page size {self._attempts - 1} times in a row without a single page succeeding, which is the configured maximum of {self._config.max_attempts} ({self._total_reductions - 1} reductions so far while reading this partition)",
+                internal_message=f"Stream {self._stream_name} reduced its page size {self._attempts - 1} times in a row without a single page succeeding, which is the configured maximum of {self._config.max_attempts} ({self._total_reductions - 1} reductions so far while reading this partition){rejection}",
                 # `transient_error`, so the only remediation is the connector's own, if it defined one.
                 message=self._with_failure_message(
                     f"The source keeps rejecting pages of stream {self._stream_name} at every page size the "
@@ -174,7 +176,7 @@ def reduce(self) -> None:
         self._current_page_size = reduced_page_size
         self._sleep(backoff)
 
-    def _retry_at_minimum_page_size(self, current_page_size: int) -> None:
+    def _retry_at_minimum_page_size(self, current_page_size: int, rejection: str) -> None:
         """
         Handle a `REDUCE_PAGE_SIZE` response that arrives when the page size is already as small as the
         connector is allowed to request.
@@ -215,7 +217,8 @@ def _retry_at_minimum_page_size(self, current_page_size: int) -> None:
                     f" ({self._retries_at_minimum_page_size} retries at that size were spent first)"
                     if self._retries_at_minimum_page_size
                     else ""
-                ),
+                )
+                + rejection,
                 message=f"The page size of stream {self._stream_name} ({current_page_size}) is already at or below "
                 f"the configured minimum of {self._config.minimum_page_size}, so the connector cannot reduce it. "
                 f"Raise the page size of the stream, or lower `minimum_page_size`.",
@@ -228,7 +231,8 @@ def _retry_at_minimum_page_size(self, current_page_size: int) -> None:
                 f", after {self._retries_at_minimum_page_size} retries at that size"
                 if self._retries_at_minimum_page_size
                 else ""
-            ),
+            )
+            + rejection,
             # `transient_error`, so the only remediation is the connector's own, if it defined one.
             message=self._with_failure_message(
                 f"The source keeps rejecting pages of stream {self._stream_name} at the smallest page size "

airbyte_cdk/sources/declarative/retrievers/simple_retriever.py

@@ -492,7 +492,7 @@ def _read_pages(
                         yield current_record
             except PaginationResetRequiredException:
                 reset_pagination = True
-            except PageSizeReductionRequiredException:
+            except PageSizeReductionRequiredException as exception:
                 if page_size_reducer is None:
                     # The action can be attached to a requester we cannot validate at config time, such as one
                     # built by a custom error handler or a CustomRequester. Re-raise as the misconfiguration it
@@ -511,7 +511,7 @@ def _read_pages(
                     )
                 # Raises once the page size cannot be reduced any further and the retries allowed at that
                 # floor are spent, which is what stops the loop when the API keeps failing.
-                page_size_reducer.reduce()
+                page_size_reducer.reduce(exception.error_message)
                 reduce_page_size = True
             else:
                 if page_size_reducer:
@@ -640,6 +640,7 @@ def _read_records_or_split_request_window(
                     f"it was rejected; splitting and re-reading the window may re-emit them."
                 )
 
+            rejection = f": {exception.error_message}" if exception.error_message else ""
             if depth >= _MAX_REQUEST_WINDOW_SPLIT_DEPTH:
                 # Logged separately from the exception below so it stays visible even if the trace message's
                 # internal_message isn't surfaced by whatever catches it.
@@ -648,13 +649,11 @@ def _read_records_or_split_request_window(
                     f"({_MAX_REQUEST_WINDOW_SPLIT_DEPTH}) while splitting {original_slice}."
                 )
                 raise AirbyteTracedException(
-                    internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}",
+                    internal_message=f"Stream {self.name} exceeded the maximum request window split depth of {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} while splitting {original_slice}; the API still rejected {stream_slice}{rejection}",
                     # `transient_error`, so the only remediation is the connector's own, if it defined one.
                     message=self._with_request_window_failure_message(
                         f"Stream {self.name} could not split its request window to a size the API accepts "
-                        f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits. This usually means the stream's "
-                        f"cursor is not actually shrinking the window on each split; if it uses a custom "
-                        f"cursor, check its `split_request_window` implementation."
+                        f"within {_MAX_REQUEST_WINDOW_SPLIT_DEPTH} splits."
                     ),
                     failure_type=FailureType.transient_error,
                 ) from exception
@@ -669,7 +668,7 @@ def _read_records_or_split_request_window(
                     else ""
                 )
                 raise AirbyteTracedException(
-                    internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further",
+                    internal_message=f"Stream {self.name} could not split its request window {stream_slice} any further; the API rejected it{rejection}",
                     # `transient_error` regardless of how the triggering response was classified: exhaustion is
                     # never the user's fault, matching `PageSizeReducer`'s equivalent exhaustion branch.
                     message=self._with_request_window_failure_message(

airbyte_cdk/sources/streams/http/page_size_reduction_exception.py

@@ -34,9 +34,10 @@ def __init__(
         detail = f": {error_message}" if error_message else ""
         super().__init__(
             internal_message=f"An error handler{stream} resolved to REDUCE_PAGE_SIZE{detail}",
-            message=f"The API rejected a page{stream} and asked the connector for a smaller one. If this message ends a sync, the stream is not set up to request a smaller page: add `page_size_reduction` to its retriever, or remove the REDUCE_PAGE_SIZE action from its error handler.",
+            message=f"The API rejected a page{stream} and asked the connector for a smaller one{detail or '.'}",
             failure_type=FailureType.config_error,
         )
+        self.error_message = error_message
 
 
 class PageSizeReductionNotSupportedException(AirbyteTracedException):

airbyte_cdk/sources/streams/http/request_window_split_exception.py

@@ -20,6 +20,10 @@ class RequestWindowSplitRequiredException(AirbyteTracedException):
     failure the built-in `HttpResponseFilter`/`ResponseAction.SPLIT_REQUEST_WINDOW` path cannot see, without
     the CDK needing to know about that connector's specific exception types.
 
+    It is raised on every split, including the ones a correctly configured stream is expected to make, so the
+    message describes the rejection and nothing else: it shows up in the traceback of any later failure of the
+    same read. A split `SimpleRetriever` cannot honor raises `RequestWindowSplitNotSupportedException` instead.
+
     The classifying error's `error_message` is preserved on this exception so it can be surfaced if the window
     is never split any further. `failure_type` is not: exhausting every window size the API will accept is
     never the user's fault, so `SimpleRetriever` always reports that as `transient_error` regardless of how the
@@ -38,9 +42,10 @@ def __init__(
         detail = f": {error_message}" if error_message else ""
         super().__init__(
             internal_message=f"A request window split was requested{stream}{detail}",
-            message=f"The API rejected the current request window{stream} and requires a smaller one. If this message ends a sync, the stream is not set up to split its window: add `request_window_splitting` to its retriever, or remove the `SPLIT_REQUEST_WINDOW` action from its error handler.",
+            message=f"The API rejected the current request window{stream} and requires a smaller one{detail or '.'}",
             failure_type=failure_type or FailureType.config_error,
         )
+        self.error_message = error_message
 
 
 class RequestWindowSplitNotSupportedException(AirbyteTracedException):

unit_tests/sources/declarative/retrievers/test_simple_retriever.py

@@ -1725,6 +1725,39 @@ def test_given_reductions_exhausted_when_read_records_then_raise_transient_error
     assert requester.send_request.call_count == 3
 
 
+@pytest.mark.parametrize(
+    "page_size_reduction",
+    [
+        pytest.param(PageSizeReduction(max_attempts=2), id="max_attempts_reached"),
+        pytest.param(PageSizeReduction(minimum_page_size=50), id="minimum_page_size_reached"),
+        pytest.param(PageSizeReduction(minimum_page_size=100), id="page_size_starts_at_minimum"),
+    ],
+)
+def test_given_reductions_exhausted_when_read_records_then_report_the_api_rejection(
+    page_size_reduction,
+):
+    """
+    The error handler's message is the only record of why the API rejected the page, so the error that ends
+    the read carries it, and its traceback does not claim the stream is not set up to reduce its page size.
+    """
+    requester = Mock(spec=Requester)
+    requester.send_request.side_effect = PageSizeReductionRequiredException(
+        stream_name=A_STREAM_NAME, error_message="Query cost 9000 exceeds 5000."
+    )
+    paginator = _mock_paginator()
+    paginator.get_page_size.return_value = 100
+    paginator.get_initial_token.return_value = None
+    retriever = _page_size_reduction_retriever(
+        requester, paginator, Mock(spec=HttpSelector), page_size_reduction
+    )
+
+    with pytest.raises(AirbyteTracedException) as exception:
+        list(retriever.read_records(A_RECORD_SCHEMA, A_STREAM_SLICE))
+
+    assert "Query cost 9000 exceeds 5000." in exception.value.internal_message
+    assert "not set up" not in exception.value.as_airbyte_message().trace.error.stack_trace
+
+
 def test_given_no_page_size_reduction_when_reduce_page_size_required_then_raise_config_error():
     requester = Mock(spec=Requester)
     requester.send_request.side_effect = PageSizeReductionRequiredException()
@@ -2647,6 +2680,64 @@ def test_given_max_split_depth_exceeded_when_read_records_then_raise_terminal_er
     assert request_window_splitter.call_count == _MAX_REQUEST_WINDOW_SPLIT_DEPTH
 
 
+def _halve_window(
+    stream_slice: StreamSlice, min_split_window: Optional[timedelta]
+) -> List[StreamSlice]:
+    start, end = stream_slice.cursor_slice["start"], stream_slice.cursor_slice["end"]
+    middle = (start + end) // 2
+    return [
+        StreamSlice(partition={}, cursor_slice={"start": start, "end": middle}),
+        StreamSlice(partition={}, cursor_slice={"start": middle + 1, "end": end}),
+    ]
+
+
+@pytest.mark.parametrize(
+    "request_window_splitter, last_window",
+    [
+        pytest.param(
+            lambda stream_slice, min_split_window: None,
+            {"start": 0, "end": 4095},
+            id="window_cannot_be_split_further",
+        ),
+        pytest.param(_halve_window, {"start": 0, "end": 3}, id="max_split_depth_reached"),
+    ],
+)
+def test_given_window_still_rejected_when_splitting_is_exhausted_then_report_the_api_rejection(
+    request_window_splitter, last_window
+):
+    """
+    The error handler's message is the only record of why the API rejected the last window, so the error that
+    ends the read carries it along with that window. A cursor that halved the window on every split is not
+    blamed for reaching the depth cap, and the traceback does not claim the stream is not set up to split.
+    """
+
+    def reject_window(**kwargs: Any) -> None:
+        raise RequestWindowSplitRequiredException(
+            stream_name=A_STREAM_NAME,
+            error_message="HTTP Status Code: 504. Error: Gateway timeout.",
+        )
+
+    requester = Mock(spec=Requester)
+    requester.send_request.side_effect = reject_window
+    paginator = _mock_paginator()
+    paginator.get_initial_token.return_value = None
+    retriever = _request_window_splitting_retriever(
+        requester, paginator, Mock(spec=HttpSelector), request_window_splitter
+    )
+
+    with pytest.raises(AirbyteTracedException) as exception:
+        list(
+            retriever.read_records(
+                A_RECORD_SCHEMA, StreamSlice(partition={}, cursor_slice={"start": 0, "end": 4095})
+            )
+        )
+
+    assert "HTTP Status Code: 504. Error: Gateway timeout." in exception.value.internal_message
+    assert str(last_window) in exception.value.internal_message
+    assert "split_request_window" not in exception.value.message
+    assert "not set up" not in exception.value.as_airbyte_message().trace.error.stack_trace
+
+
 def test_given_partitions_read_concurrently_then_window_reduction_isolates_between_partitions():
     """
     One retriever instance - and one shared `request_window_splitter` (the stream's cursor) - is used for every partition

unit_tests/sources/declarative/test_concurrent_declarative_source.py

@@ -5165,6 +5165,9 @@ def test_given_reductions_exhausted_when_read_then_emit_a_transient_error():
         "keeps rejecting pages of stream" in error.message and "records per page" in error.message
         for error in errors
     )
+    # the filter defines no `error_message`, so the default mapping's text for 502 is the rejection reported
+    assert "HTTP Status Code: 502" in errors[0].internal_message
+    assert "not set up" not in errors[0].stack_trace
 
 
 def _request_window_splitting_manifest():
@@ -5507,6 +5510,9 @@ def test_given_split_window_still_rejected_when_read_then_partition_is_not_check
     ]
     assert errors[0].failure_type == FailureType.transient_error
     assert "could not split its request window" in errors[0].internal_message
+    # the filter defines no `error_message`, so the default mapping's text for 400 is the rejection reported
+    assert "HTTP Status Code: 400" in errors[0].internal_message
+    assert "not set up" not in errors[0].stack_trace
     final_state = get_states_for_stream(stream_name="Test", messages=messages)[
         -1
     ].stream.stream_state.__dict__

unit_tests/sources/streams/http/test_http_client.py

@@ -1758,9 +1758,8 @@ def test_send_raises_page_size_reduction_required_exception_with_reduce_page_siz
     # so its message must describe the event rather than accuse the connector of a bug
     assert "should be reported" not in exception.value.message
     assert exception.value.message == (
-        "The API rejected a page of stream test and asked the connector for a smaller one. If this message "
-        "ends a sync, the stream is not set up to request a smaller page: add `page_size_reduction` to its "
-        "retriever, or remove the REDUCE_PAGE_SIZE action from its error handler."
+        "The API rejected a page of stream test and asked the connector for a smaller one: "
+        "test reduce page size message"
     )
     # a retriever that cannot re-issue the page never retries it, so a job-level retry cannot help
     assert exception.value.failure_type == FailureType.config_error
@@ -1871,9 +1870,8 @@ def test_send_raises_request_window_splitting_required_exception_with_split_requ
     assert http_client._session.send.call_count == 1
     assert "test reduce request window message" in exception.value.internal_message
     assert exception.value.message == (
-        "The API rejected the current request window of stream test and requires a smaller one. If this "
-        "message ends a sync, the stream is not set up to split its window: add `request_window_splitting` "
-        "to its retriever, or remove the `SPLIT_REQUEST_WINDOW` action from its error handler."
+        "The API rejected the current request window of stream test and requires a smaller one: "
+        "test reduce request window message"
     )
     # the classifying error's own failure_type is preserved rather than defaulting to config_error
     assert exception.value.failure_type == FailureType.config_error