feat(retries): add min_attempts and absolute_max_elapsed_time_ms to BackoffStrategy (#342)

## Summary

`retry_with_backoff_async` currently checks `now - start >
max_elapsed_time` *before* sleeping. When `RetryConfig.max_elapsed_time`
is set to e.g. 5 min and the per-attempt httpx client timeout is 30 min,
any attempt that runs longer than the budget blows it on attempt 1 —
**zero retries fire on subsequent transient errors**. Closes that
short-circuit without re-introducing the unbounded retry loops that
recent budget tightening was meant to prevent.

## What changes

New optional fields on `BackoffStrategy` (default values preserve
existing behavior):

- `min_attempts: int = 0` — minimum retry attempts that must fire before
`max_elapsed_time` is honored. Counts retries, not the initial attempt.
`min_attempts=2` permits 1 initial + at least 2 retries (3 total
attempts) before the soft cap can cut the loop.
- `absolute_max_elapsed_time_ms: int | None = None` — cap on when a new
retry can START. Does NOT interrupt an in-flight `func()` call.
Worst-case wall-clock under this cap is `absolute_max_elapsed_time_ms +
per_attempt_timeout`.

Loop changes in both sync (`retry_with_backoff`) and async
(`retry_with_backoff_async`) paths:

1. **Post-attempt cap check.** Soft cap honored only when `retries >=
min_attempts`; hard cap unconditional.
2. **Pre-sleep hard-cap check.** Refuses to sleep into a retry whose
projected start would cross the hard cap.
3. **Post-sleep verification.** Belt-and-suspenders against late wakeups
and projection rounding.
4. **Helper extraction** (`_cap_hit_after_attempt`,
`_raise_or_return_after_cap`) dedupes the soft/hard cap logic between
sync and async.

Validation in `BackoffStrategy.__init__` rejects `min_attempts < 0`,
`absolute_max_elapsed_time_ms <= 0`, and hard cap below soft cap.

`.genignore` updated to preserve these fields across future Speakeasy
regens, matching the `general.py` and `users.py` precedent.

## Design choices

- **Hard cap is a retry-start cap, not a wall-clock bound.** In-flight
`func()` cannot be interrupted from the retry loop. Consumers should
pair the cap with a sensible per-attempt timeout to keep the worst case
bounded.
- **Defaults preserve existing behavior** so this change is non-breaking
for every existing consumer. Consumers opt in by setting `min_attempts >
0` and/or `absolute_max_elapsed_time_ms`.

## Tests

49 tests pass (46 unit + 3 split-PDF retry integration). New coverage:

- **T1–T14** in `_test_unstructured_client/unit/test_retries.py`:
fake-clock harness monkeypatching `time.time` / `time.sleep` /
`asyncio.sleep` / `random.uniform`. Covers the slow-first-attempt +
`min_attempts` floor scenario, floor-is-not-a-ceiling semantics, hard
cap overrides floor, sleep truncation, `TemporaryError` early-return
through both caps, `PermanentError` short-circuit immunity, and
`BackoffStrategy.__init__` validation.
-
**`test_split_pdf_cache_tmp_data_chunk_request_stream_is_replay_safe`**
in `integration/test_decorators.py`: pins the body-replay invariant for
chunk requests built from open file objects (the
`split_pdf_cache_tmp_data=True` path). Iterates `request.stream` twice
directly — bypasses `request.read()` caching — so a future Speakeasy
template change that produced a single-consumption stream would fail
this test.

## Test plan

- [x] `uv run pytest _test_unstructured_client/unit/test_retries.py` —
46 pass
- [x] `uv run pytest
_test_unstructured_client/integration/test_decorators.py::test_split_pdf_*retry*
_test_unstructured_client/integration/test_decorators.py::test_split_pdf_cache_tmp_data_*`
— 3 pass
- [ ] CI green

Author: aballman

.genignore

@@ -19,3 +19,7 @@ src/unstructured_client/users.py
 #  - Adjust the custom url snippets in the file
 #  - Bring back the ignore line and commit
 src/unstructured_client/general.py
+
+# Custom min_attempts / absolute_max_elapsed_time_ms fields on BackoffStrategy.
+# Push upstream to Speakeasy templates to remove this entry.
+src/unstructured_client/utils/retries.py

CHANGELOG.md

@@ -3,6 +3,9 @@
 ### Breaking changes
 * Removed deprecated connector config models from the SDK (e.g. `S3SourceConnectorConfig`, `AzureDestinationConnectorConfig`). Pass connector configs as plain dicts with arbitrary fields. The SDK is no longer coupled to backend connector schemas — new fields work without an SDK upgrade.
 
+### Features
+* Add `min_attempts` and `absolute_max_elapsed_time_ms` fields to `BackoffStrategy`. `min_attempts` is the minimum number of retry attempts that must fire before `max_elapsed_time` is honored; defaults to `0` (preserves existing behavior). `absolute_max_elapsed_time_ms` caps when a new retry can start (does not interrupt in-flight requests); defaults to `None`. Together these close a short-circuit where a single slow first attempt could exhaust the retry budget before any retry fired. See FS-1988.
+
 ## 0.43.4
 
 ### Enhancements

_test_unstructured_client/integration/test_decorators.py

@@ -813,3 +813,45 @@ async def mock_send(_, request: httpx.Request, **kwargs):
     assert number_of_transport_failures == 0
     assert mock_endpoint_called
     assert res.status_code == 200
+
+
+def test_split_pdf_cache_tmp_data_chunk_request_stream_is_replay_safe(tmp_path):
+    from unstructured_client._hooks.custom.request_utils import (
+        create_pdf_chunk_request,
+    )
+
+    chunk_path = tmp_path / "chunk.pdf"
+    src_bytes = Path("_sample_docs/layout-parser-paper.pdf").read_bytes()
+    chunk_path.write_bytes(src_bytes)
+
+    pdf_chunk_file = open(chunk_path, "rb")  # noqa: SIM115
+    try:
+        form_data = {
+            "files": (chunk_path.name, src_bytes, "application/pdf"),
+            "strategy": "fast",
+        }
+        original_request = httpx.Request(
+            method="POST",
+            url="http://localhost:8000/general/v0/general",
+            headers={
+                "Content-Type": "multipart/form-data; boundary=test",
+                "User-Agent": "test",
+            },
+            content=b"",
+        )
+
+        chunk_request = create_pdf_chunk_request(
+            form_data=form_data,
+            pdf_chunk=(pdf_chunk_file, 1),
+            original_request=original_request,
+            filename=chunk_path.name,
+        )
+
+        # Iterate twice without request.read() to bypass _content caching.
+        first_pass = b"".join(chunk_request.stream)
+        second_pass = b"".join(chunk_request.stream)
+
+        assert len(first_pass) > 1000
+        assert first_pass == second_pass
+    finally:
+        pdf_chunk_file.close()