fix: Stop dataset iterators from skipping items when unwind is used (#1059)

## What was wrong

`iterate_items` advanced the dataset offset by `max(scanned_rows,
len(items))`. With `unwind`, a page returns more
items than the rows it scanned, so the offset jumped past rows the next
request never read. At the default
`chunk_size`, a 5000-row dataset with a 3x unwind yielded about a third
of it.

## The fix

The offset now advances by at most the number of rows the call asked
for. The items endpoint applies `offset` and
`limit` to the rows first and shapes the result afterwards, and it
passes no `maxLimit` (unlike the collection list
routes), so a page never covers more rows than the `limit` it was sent.
Capping the advance there keeps an unwound
page from running past the window it actually read.

The `max()` stays underneath the cap on purpose.
`x-apify-pagination-count` comes from the dataset's `itemCount`,
which the API increments through a throttled write while the items
themselves stream fresh. Following the header
alone, which is what
[apify-client-js#1044](apify/apify-client-js#1044)
does and what #1058
suggested, would truncate `iterate_items` right after `push_items` for
anything over one page. The JS iterator
bounds itself by `total` and so is exposed to that lag either way; this
one consults neither, which is why the two
clients end up with different fixes.

`DatasetItemsPage.count` keeps its current value of `max(header,
len(items))`. That is not the no-op it looks like:
it is what makes `count` usable right after a push, and two integration
assertions depend on it. Its docstring
changes, along with the `limit` docs on `iterate_items` and `chunk_size`
on both twins, since all of them promised
items where the field and the parameters have always counted scanned
rows.

One case stays imperfect. On a final short page with `unwind`, the
advance is still the full `limit` rather than the
rows the page covered, so rows appended by a concurrent push after that
page can be missed. Reaching them needs the
raw header, which `list_items` folds into `count`. It is strictly better
than before, and the docstring says so
rather than claiming the overshoot is free.

## Behavior change worth a changelog line

`iterate_items(limit=N)` with `unwind` and a `chunk_size` below `N` now
walks all N rows instead of stopping after
the first page, so it can yield several times more items for the same
`limit`. That is the fix working, but it is
user-visible.

## Tests

- `unwind` reached the pagination tests for the first time. The fake API
now splits each row into three items after
the row window is picked, exactly as the transform stream does, and
three cases iterate through it.
- The fake API used to cap every page at 1000 items "mirroring the real
API", which is only true of the collection
endpoints. It now applies the requested limit verbatim on the items
endpoint, which is what let a `chunk_size`
  above that cap be covered at all.
- A page whose `count` lags behind its items has a test of its own.
Nothing pinned that before, and it is the
  invariant the `max()` rests on.
- `test_dataset_iterate_items_unwound` covers the whole thing against
the live API. It has not run locally, so CI
  is its first real execution.

## One thing found while verifying

`get_cursor_iterator` justified its termination rule with filters that
"can drop every item on a page while a live
cursor still points at more data". The API rules that out: the key-value
store returns the last key of the page as
the next cursor, and the request queue returns a cursor only once a page
came back full, so neither can hand back a
cursor that outlives its page. The behavior is fine and unchanged; the
docstring now says why it holds.

Closes #1058

*✍️ Drafted by Claude Code*

Author: vdusek

docs/02_concepts/08_pagination.mdx

@@ -25,6 +25,8 @@ Most methods named `list` or `list_something` in the Apify client return a page
 - `count` - The number of items in the current page.
 - `limit` - The maximum number of items per page.
 
+On <ApiLink to="class/DatasetItemsPage">`DatasetItemsPage`</ApiLink>, `count` reports how many dataset rows the API scanned for the page. Filters drop items from a page and `unwind` multiplies them, so a page can hold fewer or more items than `count`.
+
 Some methods paginate differently. For example, <ApiLink to="class/RequestQueueClient#list_requests">`RequestQueueClient.list_requests`</ApiLink> returns a cursor-based <ApiLink to="class/ListOfRequests">`ListOfRequests`</ApiLink> without the `total`, `offset`, and `count` fields. To fetch the next page, pass its `next_cursor` value back as the `cursor` parameter. Other examples include `list_keys` and `list_head`. Regardless, the primary results are always stored under the `items` field, and the `limit` field can be used to control the number of results returned.
 
 The following example shows how to fetch all items from a dataset using pagination:

src/apify_client/_pagination.py

@@ -19,9 +19,10 @@
 class HasItems(Protocol[T]):
     """Structural contract for a single page of results from a paginated API endpoint.
 
-    Implementations must expose `items`. They may optionally expose `count` - the number of items scanned by the API for
-    this page, which can exceed `len(items)` when filters drop items from the response. The iterator helpers consult
-    `count` opportunistically via `getattr` for offset bookkeeping and fall back to `len(items)` when it is absent.
+    Implementations must expose `items`. They may optionally expose `count` - the number of rows the API scanned to
+    produce this page, which `len(items)` can land below (filters drop items) or above (`unwind` splits one row into
+    several items). The iterator helpers consult `count` opportunistically via `getattr` for offset bookkeeping and
+    fall back to `len(items)` when it is absent.
     """
 
     items: list[T]
@@ -38,34 +39,34 @@ def get_items_iterator(
 
     The `callback` is invoked lazily to fetch each page from the API. It must accept `limit` and `offset` keyword
     arguments and return an object whose `items` attribute is a list. If the object also exposes a `count` attribute, it
-    is used for offset bookkeeping (the Apify API's `count` reflects items scanned, which can exceed items returned when
-    filters are applied).
+    is used for offset bookkeeping - `_page_scanned_rows` describes how the next offset is derived.
 
-    Iteration stops when a page scans no items (`count` is `0`, or `items` is empty when `count` is absent) or when the
-    user-requested `limit` is reached. A page can scan items while returning none - filters like `clean` drop items from
-    `items` but still count toward `count` - so terminating on scanned rather than returned items keeps the iterator
-    advancing across fully-filtered pages. The `total` field is intentionally not consulted, because it can change
-    between calls.
+    Iteration stops when a page scans no rows or when the user-requested `limit` is reached. A page can scan rows while
+    returning no items - filters like `clean` drop items from `items` but still count toward `count` - so terminating on
+    scanned rather than returned rows keeps the iterator advancing across fully-filtered pages. The `total` field is
+    intentionally not consulted, because it can change between calls.
 
     Args:
         callback: Function returning a single page of items.
-        limit: Maximum total number of items to yield across all pages. `None` or `0` means no limit.
+        limit: Maximum total number of rows scanned across all pages. On the dataset items endpoint `unwind` can
+            turn one row into several items, so more items than this can be yielded. `None` or `0` means no limit.
         offset: Starting offset for the first page.
-        chunk_size: Maximum number of items requested per API call. `None` or `0` lets the API decide.
+        chunk_size: Per-page cap, sent to the API as its `limit`. `None` or `0` lets the API decide.
     """
     effective_chunk = chunk_size or 0
     initial_offset = offset or 0
     initial_limit = limit or 0
     fetched_items = 0
 
     while True:
+        page_limit = _next_page_limit(initial_limit, fetched_items, effective_chunk)
         current_page = callback(
-            limit=_next_page_limit(initial_limit, fetched_items, effective_chunk),
+            limit=page_limit,
             offset=initial_offset + fetched_items,
         )
         yield from current_page.items
 
-        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = _page_scanned_rows(current_page, page_limit)
         fetched_items += page_scanned
 
         if not page_scanned or (initial_limit and fetched_items >= initial_limit):
@@ -89,14 +90,15 @@ async def get_items_iterator_async(
     fetched_items = 0
 
     while True:
+        page_limit = _next_page_limit(initial_limit, fetched_items, effective_chunk)
         current_page = await callback(
-            limit=_next_page_limit(initial_limit, fetched_items, effective_chunk),
+            limit=page_limit,
             offset=initial_offset + fetched_items,
         )
         for item in current_page.items:
             yield item
 
-        page_scanned = max(getattr(current_page, 'count', 0), len(current_page.items))
+        page_scanned = _page_scanned_rows(current_page, page_limit)
         fetched_items += page_scanned
 
         if not page_scanned or (initial_limit and fetched_items >= initial_limit):
@@ -126,20 +128,30 @@ def get_cursor_iterator(
     limit: int | None = None,
     chunk_size: int | None = None,
 ) -> Iterator[KeyValueStoreKey] | Iterator[Request]:
-    """Yield individual items from cursor-paginated API responses.
+    """Yield individual items from a cursor-paginated API response.
+
+    This iterator supports the two API responses that use cursor pagination. `ListOfKeys` is used for key-value store
+    keys, while `ListOfRequests` is used for request queue requests.
+
+    Pagination continues until either:
+
+    - the API returns no next cursor, or
+    - the requested `limit` is reached.
+
+    An empty page does not explicitly stop the iteration. In practice, both supported endpoints return a next cursor
+    only when the current page contains items, so an empty page always has a `None` cursor and naturally ends the
+    iteration.
 
-    Cursor pagination is restricted to the two API responses that expose it: `ListOfKeys` (for key-value store keys) and
-    `ListOfRequests` (for request queue requests). Iteration ends when the next cursor is `None` or the user-requested
-    `limit` is reached. Emptiness alone does not stop iteration: server-side filters (such as the request-queue state
-    `filter`) can drop every item on a page while a live cursor still points at more data, so termination relies on the
-    cursor, not on whether a page returned items. Unlike offset responses, cursor responses expose no scanned-item
-    `count`, so `count` cannot be used to detect a fully-filtered page here.
+    The endpoints determine the next cursor differently:
+
+    - For key-value store keys, the cursor is the last key returned on the current page.
+    - For request queue requests, a cursor is returned only when the current page is full.
 
     Args:
-        callback: Function returning a single page of items. Receives `cursor` and `limit` kwargs.
-        cursor: Value of the cursor for the first request, or `None` to start from the beginning.
+        callback: Function that returns one page of items and accepts `cursor` and `limit` keyword arguments.
+        cursor: Cursor to use for the first request. If `None`, iteration starts from the beginning.
         limit: Maximum total number of items to yield across all pages.
-        chunk_size: Maximum number of items requested per API call.
+        chunk_size: Maximum number of items to request in a single API call.
     """
     effective_chunk = chunk_size or 0
     initial_limit = limit or 0
@@ -218,3 +230,19 @@ def _next_page_limit(initial_limit: int, fetched_items: int, effective_chunk: in
     if not effective_chunk:
         return remaining
     return min(remaining, effective_chunk)
+
+
+def _page_scanned_rows(page: HasItems[T], requested_limit: int) -> int:
+    """Compute how far the offset advances past `page`, in dataset rows.
+
+    Neither reported number is right on its own. `count` follows the rows the API scanned, but it is derived from a
+    dataset's item count, which is incremented by a throttled write and so lags a fresh push. `len(items)` counts the
+    items the API shaped out of those rows: filters (`clean`, `skip_empty`, `skip_hidden`) drop some, and `unwind`
+    splits one row into several. The larger of the two absorbs a `count` that lags behind the items returned, and
+    capping it at the rows the call asked for keeps an unwound page from advancing past rows the next call would then
+    never read. The cap is a valid bound because the endpoint applies the `limit` it is sent verbatim; on a page
+    covering fewer rows than that, the advance can still overshoot into rows a concurrent push appends afterwards. A
+    `requested_limit` of `0` means the call sent no limit, leaving the advance unbounded.
+    """
+    scanned_rows = max(getattr(page, 'count', 0), len(page.items))
+    return min(scanned_rows, requested_limit) if requested_limit else scanned_rows

src/apify_client/_resource_clients/dataset.py

@@ -40,7 +40,7 @@ class DatasetItemsPage:
     """The offset of the first item in this page."""
 
     count: int
-    """Number of items in this page."""
+    """Number of dataset rows the API scanned for this page, or the number of items returned when that is larger."""
 
     limit: int
     """The limit that was used for this request."""
@@ -204,8 +204,8 @@ def list_items(
             items=items,
             total=int(response.headers['x-apify-pagination-total']),
             offset=int(response.headers['x-apify-pagination-offset']),
-            # x-apify-pagination-count returns count of processed items, not count of returned items
-            # This makes difference when items were filtered using hidden/empty
+            # The header counts the rows the API scanned, which `unwind` and a lagging dataset item count can
+            # both leave below the number of items returned.
             count=max(int(response.headers['x-apify-pagination-count']), len(items)),
             # API returns 999999999999 when no limit is used
             limit=int(response.headers['x-apify-pagination-limit']),
@@ -237,7 +237,8 @@ def iterate_items(
 
         Args:
             offset: Number of items that should be skipped at the start. The default value is 0.
-            limit: Maximum number of items to return. By default there is no limit.
+            limit: Maximum number of dataset rows to scan. Fewer items are yielded when filters drop some, more
+                when `unwind` splits a row into several. By default there is no limit.
             desc: By default, results are returned in the same order as they were stored. To reverse the order,
                 set this parameter to True.
             clean: If True, returns only non-empty items and skips hidden fields (i.e. fields starting with
@@ -260,7 +261,7 @@ def iterate_items(
             skip_hidden: If True, then hidden fields are skipped from the output, i.e. fields starting with
                 the # character.
             signature: Signature used to access the items.
-            chunk_size: Maximum number of items requested per API call when iterating across pages.
+            chunk_size: Maximum number of dataset rows requested per API call when iterating across pages.
             timeout: Timeout for the API HTTP request.
 
         Yields:
@@ -763,8 +764,8 @@ async def list_items(
             items=items,
             total=int(response.headers['x-apify-pagination-total']),
             offset=int(response.headers['x-apify-pagination-offset']),
-            # x-apify-pagination-count returns count of processed items, not count of returned items
-            # This makes difference when items were filtered using hidden/empty
+            # The header counts the rows the API scanned, which `unwind` and a lagging dataset item count can
+            # both leave below the number of items returned.
             count=max(int(response.headers['x-apify-pagination-count']), len(items)),
             # API returns 999999999999 when no limit is used
             limit=int(response.headers['x-apify-pagination-limit']),
@@ -796,7 +797,8 @@ def iterate_items(
 
         Args:
             offset: Number of items that should be skipped at the start. The default value is 0.
-            limit: Maximum number of items to return. By default there is no limit.
+            limit: Maximum number of dataset rows to scan. Fewer items are yielded when filters drop some, more
+                when `unwind` splits a row into several. By default there is no limit.
             desc: By default, results are returned in the same order as they were stored. To reverse the order,
                 set this parameter to True.
             clean: If True, returns only non-empty items and skips hidden fields (i.e. fields starting with
@@ -819,7 +821,7 @@ def iterate_items(
             skip_hidden: If True, then hidden fields are skipped from the output, i.e. fields starting with
                 the # character.
             signature: Signature used to access the items.
-            chunk_size: Maximum number of items requested per API call when iterating across pages.
+            chunk_size: Maximum number of dataset rows requested per API call when iterating across pages.
             timeout: Timeout for the API HTTP request.
 
         Yields:

tests/integration/test_dataset.py

@@ -596,6 +596,47 @@ async def get_items() -> DatasetItemsPage:
         await maybe_await(dataset_client.delete())
 
 
+async def test_dataset_iterate_items_unwound(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None:
+    """Test iterate_items with `unwind`, where a page carries more items than the rows it scanned."""
+    dataset_name = get_random_resource_name('dataset')
+    created_dataset = await maybe_await(client.datasets().get_or_create(name=dataset_name))
+    assert isinstance(created_dataset, Dataset)
+    dataset_client = client.dataset(created_dataset.id)
+
+    try:
+        items_to_push = [{'idx': i, 'parts': [{'part': p} for p in range(3)]} for i in range(12)]
+        await maybe_await(dataset_client.push_items(items_to_push))
+
+        # Poll until all 12 rows are visible (eventual consistency) so the chunked iteration sees every page
+        async def get_items() -> DatasetItemsPage:
+            page = await maybe_await(dataset_client.list_items(limit=12))
+            assert isinstance(page, DatasetItemsPage)
+            return page
+
+        await poll_until_condition(get_items, lambda page: len(page.items) == 12)
+
+        # chunk_size=5 caps a page at 5 rows, which `unwind` expands into 15 items
+        iterator = dataset_client.iterate_items(unwind=['parts'], chunk_size=5)
+        collected: list[dict] = []
+        if is_async:
+            assert isinstance(iterator, AsyncIterator)
+            async for item in iterator:
+                assert isinstance(item, dict)
+                collected.append(item)
+        else:
+            assert isinstance(iterator, Iterator)
+            for item in iterator:
+                assert isinstance(item, dict)
+                collected.append(item)
+
+        # Every part of every row arrives exactly once: no page is skipped and none is read twice.
+        assert sorted((item['idx'], item['part']) for item in collected) == [
+            (idx, part) for idx in range(12) for part in range(3)
+        ]
+    finally:
+        await maybe_await(dataset_client.delete())
+
+
 async def test_dataset_iterate_items_with_fields(client: ApifyClient | ApifyClientAsync, *, is_async: bool) -> None:
     """Test iterate_items with `fields` filter."""
     dataset_name = get_random_resource_name('dataset')