http/concurrent: no stall detection in the declarative HTTP path (no request timeout, unused timeout_seconds, no repeated-page-token guard)

## Symptom

A declarative (manifest) source whose HTTP response connects and then stalls, or whose cursor pagination stops advancing, emits neither RECORD nor STATE for the whole duration of the stall. Nothing in the CDK ever interrupts it. The attempt ends only when the platform source heartbeat fires (`heartbeat-max-seconds-between-messages`, documented default 3 hours: https://github.com/airbytehq/airbyte/blob/master/docs/platform/understanding-airbyte/heartbeats.md; the Cloud setting observed in the incident below was 5400 s). Because the partition never closes, no state is checkpointed, and every retry repeats the identical walk from the last committed state and stalls at the same point.

Observed in production on source-zendesk-support `tickets` (Incremental Ticket Export `incremental/tickets/cursor.json`, `CursorPagination` on `after_url` with `stop_condition` keyed on `end_of_stream`, `RequestPath` token option): with `num_workers` 1, or once sibling streams finish, the source went silent until the heartbeat timeout. Internal reference: airbytehq/oncall#13250.

## Root cause

Three independent layers, none of which bounds a hung request or a non-advancing pagination loop.

**1. `HttpClient` sets no request timeout.**
The file contains no occurrence of the string `timeout` at either ref. `_send` forwards `request_kwargs` verbatim to `requests.Session.send`:
- v7.23.8: `_send` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/streams/http/http_client.py#L325, `self._session.send(request, **request_kwargs)` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/streams/http/http_client.py#L348
- main: `_send` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/streams/http/http_client.py#L391, `self._session.send(...)` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/streams/http/http_client.py#L414

`HttpRequester.send_request` only passes `{"stream": self.stream_response}`:
- v7.23.8: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/requesters/http_requester.py#L466 (field: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/requesters/http_requester.py#L70)
- main: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/requesters/http_requester.py#L466 (field: https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/requesters/http_requester.py#L70)

`requests` defaults to `timeout=None` (https://requests.readthedocs.io/en/latest/user/advanced/#timeouts), so a server that completes the TCP/TLS handshake, sends headers, and then stops sending bytes blocks the worker thread indefinitely. With `stream=False` the block is inside `_send`; with `stream=True` it moves to the decoder iterating the body. In both cases the worker is wedged and the error handler / backoff never runs.

**2. `ConcurrentSource.timeout_seconds` is accepted, documented, stored, and never read.**
- `DEFAULT_TIMEOUT_SECONDS = 900`: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L38 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L38
- Docstring promising "If no record is read within this time, the source will stop reading and return": https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L93 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L93
- The only occurrence of `_timeout_seconds` in the repository at either ref is the assignment (`git grep _timeout_seconds` returns one line); nothing reads it: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L100 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L100
- `_consume_from_queue` blocks on `queue.get()` with no timeout: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L151 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/concurrent_source/concurrent_source.py#L151
- `ConcurrentDeclarativeSource` does not pass it either: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L246 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/concurrent_declarative_source.py#L265

The CDK therefore has no independent watchdog for a partition whose worker never returns. The main thread waits forever on the queue.

**3. `SimpleRetriever._read_pages` has no guard against a repeated page token.**
The loop is `while True`; its only exits are a falsy response or the paginator returning `None`:
- v7.23.8: `_read_pages` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L343, `while True` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L352, `if not response: break` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L407, `if not next_page_token: break` https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L430
- main: `_read_pages` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L349, `while True` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L358, `if not response: break` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L413, `if not next_page_token: break` https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L436

The pagination-reset branch does not help: `PaginationTracker.has_reached_limit` is record-count based, so a sequence of empty pages never triggers it (https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/pagination_tracker.py#L44 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/pagination_tracker.py#L44).

The previous token is already computed and handed to the paginator (`last_page_token_value`: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L421 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/retrievers/simple_retriever.py#L427), but `CursorPaginationStrategy.next_page_token` accepts it and never compares against it; it only evaluates `stop_condition` and returns `cursor_value` as-is:
- parameter: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L79 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L79
- `stop_condition` eval: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L86 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L86
- `return token if token else None`: https://github.com/airbytehq/airbyte-python-cdk/blob/v7.23.8/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L103 / https://github.com/airbytehq/airbyte-python-cdk/blob/main/airbyte_cdk/sources/declarative/requesters/paginators/strategies/cursor_pagination_strategy.py#L103

So an API that keeps returning the same `after_url` with `end_of_stream: false` and an empty page makes `_read_pages` re-request the same URL forever, emitting nothing.

## Reproduction

Any manifest-only connector against a local stub server; `airbyte-cdk connector test` or `poetry run source-declarative-manifest read ...` with `concurrency_level` 1.

A. Stalled response (layers 1 and 2):
1. Run a stub that returns `200` with headers and then never sends a body:
```python
from http.server import BaseHTTPRequestHandler, HTTPServer
import time
class H(BaseHTTPRequestHandler):
def do_GET(self):
self.send_response(200); self.send_header("Content-Type", "application/json"); self.end_headers()
time.sleep(10**6)
HTTPServer(("127.0.0.1", 8080), H).serve_forever()
```
2. Point a single-stream manifest (`HttpRequester`, `url_base: http://127.0.0.1:8080`) at it and run `read`.
3. Expected: a timeout error after a bounded interval, routed through the error handler, or the source giving up after `timeout_seconds` (900 s) as the docstring states.
4. Observed: no RECORD, no STATE, no log line; the process never terminates. `timeout_seconds` has no effect at any value.

B. Non-advancing cursor (layer 3):
1. Stub always answers `{"tickets": [], "after_url": "http://127.0.0.1:8080/tickets?cursor=x", "end_of_stream": false}`.
2. Manifest paginator:
```yaml
paginator:
type: DefaultPaginator
page_token_option: { type: RequestPath }
pagination_strategy:
type: CursorPagination
cursor_value: "{{ response.after_url }}"
stop_condition: "{{ response.end_of_stream }}"
```
3. Expected: the retriever stops with an error once the token repeats.
4. Observed: identical request re-issued indefinitely; no records, no state, no error.

## Impact

- Affects every declarative connector: `HttpClient` and `SimpleRetriever` are the shared path for all manifest-only and low-code sources.
- Failure mode is silent: no output, no error, no diagnostic. Only the platform heartbeat ends the attempt, so each attempt burns the whole heartbeat window (90 minutes at the Cloud setting seen in the incident, 3 hours at the documented default) before failing.
- No state is checkpointed because the partition never closes, so retries do not progress and the sync cannot self-heal.
- The connector-side mitigation in airbytehq/airbyte#85760 (raise the minimum concurrent threads to 2 so a sibling stream keeps the heartbeat alive) only delays the failure; once siblings finish, the stalled partition alone remains and the heartbeat fires anyway.
- `timeout_seconds` gives connector authors a false sense of protection: it is accepted and documented but inert.

## Suggested fix

Three independent changes, smallest first. (1) alone resolves the observed incident.

1. **Default read timeout in `HttpClient._send`.** If `"timeout"` is absent from `request_kwargs`, set a default such as `(30, 600)` (connect, read) before `self._session.send(...)` (v7.23.8 L348 / main L414). Expose an override on `HttpRequester` (a `request_timeout` field next to `stream_response`, L70) merged into `request_kwargs` at L466. A stalled non-streamed response then raises `requests.exceptions.ReadTimeout` inside `_send`, which already catches `RequestException` and hands it to `self._error_handler.interpret_response` (`except` at v7.23.8 L349 / main L415, `interpret_response` at v7.23.8 L352 / main L421); a stalled streamed body raises `requests.exceptions.ConnectionError` from `iter_content` (requests wraps urllib3 `ReadTimeoutError` there), which still un-wedges the worker and fails the partition with a real error. Read timeout is per-byte-gap, not total, so large legitimate downloads are unaffected.

2. **Wire `timeout_seconds` into `_consume_from_queue`.** Replace `queue.get()` (L151) with `queue.get(timeout=self._timeout_seconds)`; on `queue.Empty`, log the partitions still in flight and raise so the attempt fails with a diagnostic instead of waiting for the platform heartbeat. Pass `timeout_seconds` from `ConcurrentDeclarativeSource` (L246 / L265) so it is configurable. Keep 900 s as default and ensure the HTTP read timeout from (1) is shorter, so the HTTP layer fires first and this becomes a last-resort watchdog.

3. **Optional: repeated-token guard in `_read_pages`.** After `_next_page_token` (L424 / L430), if the new token equals the previous `last_page_token_value` and `last_page_size == 0`, log an error naming the stream, slice, and token, and break (or raise) instead of re-requesting. The previous token is already available at that point (L421 / L427), so this is a few lines with no signature change. Logging rather than silently returning `None` from the strategy keeps the failure visible.

## Precedent

- airbytehq/airbyte#85760 "fix(source-zendesk-support): raise minimum concurrent threads to 2 and migrate existing configs". Review of that PR surfaced this gap: the fix keeps the heartbeat alive via sibling streams but cannot terminate the stalled partition itself.
- Internal incident reference: airbytehq/oncall#13250 (private).
- Platform heartbeat contract: https://github.com/airbytehq/airbyte/blob/master/docs/platform/understanding-airbyte/heartbeats.md (RECORD and STATE count as heartbeats; default window 3 hours).
- The `timeout_seconds` docstring (L93) already describes the intended behavior ("the source will stop reading and return"); this issue asks for that contract to be implemented.

Start by reading HttpClient._send, HttpRequester.send_request, ConcurrentSource._consume_from_queue, and SimpleRetriever._read_pages with CursorPaginationStrategy.next_page_token. Reproduce both cases using the local stub-server examples and the declarative manifest read command. Done means stalled responses and repeated cursor tokens terminate with a bounded, observable failure rather than emitting nothing or looping indefinitely.

Scritto dal modello di indicizzazione a partire dal testo della issue.