fix!: keep pagination iterators advancing past fully filtered pages

docs/04_upgrading/upgrading_v3.md

@@ -272,6 +272,22 @@ A URL field whose value isn't a valid absolute URL now fails response validation
 
 <ApiLink to="class/ActorVersionCollectionClient#list">`ActorVersionCollectionClient.list()`</ApiLink> and <ApiLink to="class/ActorEnvVarCollectionClient#list">`ActorEnvVarCollectionClient.list()`</ApiLink> now take no arguments. Neither endpoint reads `offset`, `limit` or `desc`, and both return every item in one response, so `chunkSize` had nothing to size either. The `ActorVersionCollectionListOptions` and `ActorEnvVarCollectionListOptions` types that declared those four options, deprecated since v2.21.0, are gone from the package. A call that passed an options object no longer compiles. Drop the argument and the call returns the same items as before.
 
+## `limit` counts scanned rows when iterating `listItems()`
+
+Iterating <ApiLink to="class/DatasetClient#listItems">`DatasetClient.listItems()`</ApiLink> with `for await` pages through the dataset by `offset` and `limit`. The API applies both to the dataset's rows before `clean`, `skipEmpty`, `skipHidden` and `unwind` reshape them, so a page can return fewer items than the rows it covered, or more. v2 advanced the offset and counted down `limit` by the items each page returned. A filter made the next page re-read rows the previous one had covered, `unwind` made it skip rows, and a page that came back empty ended the loop even when rows remained behind it.
+
+v3 advances and stops by the number of rows the API scanned, taken from the `x-apify-pagination-count` header. `limit` therefore caps the rows scanned, the same way it does on a single awaited `listItems()` call and in the API itself. With a filter set, the loop can yield fewer than `limit` items, each of them once. In v2 the same call made up the difference with items it had already yielded, or stopped early. With `unwind`, the loop can yield more than `limit` items. To collect a fixed number of items from a filtered dataset, count them in the loop:
+
+```js
+const items = [];
+for await (const item of client.dataset('dataset-id').listItems({ clean: true, chunkSize: 100 })) {
+    items.push(item);
+    if (items.length === 100) break;
+}
+```
+
+`PaginatedList.count` keeps reporting the items returned on the page. The cursor-based iterators, `listKeys()`, `listRequests()` and `paginateRequests()`, are unchanged.
+
 ## The last deprecated options are gone
 
 Two options that carried a `@deprecated` marker throughout v2 have been removed.

src/base/api_client.ts

@@ -1,7 +1,7 @@
 import type { ApifyClient } from '../apify_client.js';
 import type { HttpClient } from '../http_client.js';
 import type { PaginatedResponse, PaginationOptions } from '../utils.js';
-import { toPath, toPathSegment } from '../utils.js';
+import { SCANNED_COUNT, toPath, toPathSegment } from '../utils.js';
 
 /** @private */
 export interface ApiClientOptions {
@@ -112,17 +112,25 @@ export abstract class ApiClient {
             limit: minForLimitParam(options.limit, chunkSize),
         } as T);
 
+        // A page can return more or fewer items than the rows it scanned (see `SCANNED_COUNT`). The next offset and
+        // the stop condition follow the scanned number alone: advancing by `items.length` would re-scan rows after a
+        // filter dropped some and skip rows after `unwind` multiplied them, and stopping at an empty page would end
+        // the iteration in front of rows a filter hid.
+        const scannedRows = (page: R): number =>
+            (page as { [SCANNED_COUNT]?: number })[SCANNED_COUNT] ?? page.items.length;
+
         async function* asyncGenerator() {
             let currentPage = await paginatedListPromise;
             yield* currentPage.items;
             const offset = options.offset ?? 0;
             const limit = Math.min(options.limit || currentPage.total, currentPage.total);
 
-            let currentOffset = offset + currentPage.items.length;
-            let remainingItems = Math.min(currentPage.total - offset, limit) - currentPage.items.length;
+            let pageScanned = scannedRows(currentPage);
+            let currentOffset = offset + pageScanned;
+            let remainingItems = Math.min(currentPage.total - offset, limit) - pageScanned;
 
             while (
-                currentPage.items.length > 0 && // Continue only if at least some items were returned in the last page.
+                pageScanned > 0 && // Continue only if the last page scanned some rows.
                 remainingItems > 0
             ) {
                 const newOptions = {
@@ -132,8 +140,9 @@ export abstract class ApiClient {
                 } as T;
                 currentPage = await getPaginatedList(newOptions);
                 yield* currentPage.items;
-                currentOffset += currentPage.items.length;
-                remainingItems -= currentPage.items.length;
+                pageScanned = scannedRows(currentPage);
+                currentOffset += pageScanned;
+                remainingItems -= pageScanned;
             }
         }
 

src/resource_clients/dataset.ts

@@ -22,6 +22,7 @@ import {
     paginationOptionsShape,
     parseArgument,
     parseResponse,
+    SCANNED_COUNT,
 } from '../utils.js';
 
 // A predicate, not a `z.object()` arm, which would walk and copy every key of every pushed item.
@@ -403,15 +404,22 @@ export class DatasetClient<
     }
 
     private _createPaginationList(response: ApifyResponse, userProvidedDesc: boolean): PaginatedList<Data> {
-        return {
+        const page: PaginatedList<Data> = {
             items: response.data,
             total: Number(response.headers['x-apify-pagination-total']),
             offset: Number(response.headers['x-apify-pagination-offset']),
-            count: response.data.length, // because x-apify-pagination-count returns invalid values when hidden/empty items are skipped
+            // `x-apify-pagination-count` reports the rows scanned, which filters and `unwind` can leave above or below
+            // the items returned.
+            count: response.data.length,
             limit: Number(response.headers['x-apify-pagination-limit']), // API returns 999999999999 when no limit is used
             // TODO: Replace this once https://github.com/apify/apify-core/issues/3503 is solved
             desc: JSON.parse(response.headers['x-apify-pagination-desc'] ?? userProvidedDesc),
         };
+
+        // The offset iterator paginates by the scanned number, so it travels with the page outside its public shape.
+        return Object.defineProperty(page, SCANNED_COUNT, {
+            value: Number(response.headers['x-apify-pagination-count']) || 0,
+        });
     }
 }
 

src/utils.ts

@@ -490,6 +490,17 @@ export interface PaginatedList<Data> extends PaginatedResponse<Data> {
     desc: boolean;
 }
 
+/**
+ * Key under which a page of dataset items carries the number of rows the API scanned to produce it, as a
+ * non-enumerable property that stays out of the page's public shape. The API applies `offset` and `limit` to the
+ * dataset's rows first and shapes the result afterwards: filters (`clean`, `skipEmpty`, `skipHidden`) drop items and
+ * `unwind` splits or drops them, so `items.length` can land on either side of the rows scanned. The
+ * `x-apify-pagination-count` header reports the scanned number, and the offset iterator advances and terminates by it
+ * alone. A page without the property falls back to `items.length`.
+ * @internal
+ */
+export const SCANNED_COUNT = Symbol('scannedCount');
+
 /**
  * Type representing both a Promise of a paginated list and an async iterable.
  *

test/datasets.test.ts

@@ -157,7 +157,7 @@ describe('Dataset methods', () => {
                 'content-type': 'application/json; chartset=utf-8',
                 'x-apify-pagination-total': '0',
                 'x-apify-pagination-offset': '0',
-                'x-apify-pagination-count': '1', // wrong on purpose to check that it's not used
+                'x-apify-pagination-count': '1', // the scanned rows; `count` reports the items returned, so it stays 0
                 'x-apify-pagination-limit': '100000',
                 // TODO: https://github.com/apify/apify-core/issues/3503
                 'x-apify-pagination-desc': false,

test/pagination.test.ts

@@ -302,6 +302,78 @@ describe('DatasetClient.listItems as async iterable', () => {
             mockedClient.mockRestore();
         }
     } as any);
+
+    // Mimics the API on a dataset whose rows are reshaped before they are returned: `offset` and `limit` pick the
+    // scanned window over all rows first, `transform` (a filter, an `unwind`) then decides what gets returned, and
+    // `x-apify-pagination-count` still reports the whole window.
+    const mockTransformedDataset = (totalRows: number, transform: (rows: ReturnType<typeof range>) => unknown[]) =>
+        vi.spyOn(client.httpClient, 'call').mockImplementation((async (request: any) => {
+            const offset = request.params.offset ?? 0;
+            const scanned = range(offset, Math.min(offset + (request.params.limit || totalRows), totalRows));
+
+            return {
+                data: transform(scanned),
+                headers: {
+                    'x-apify-pagination-total': String(totalRows),
+                    'x-apify-pagination-offset': String(offset),
+                    'x-apify-pagination-count': String(scanned.length),
+                    'x-apify-pagination-limit': String(request.params.limit),
+                    'x-apify-pagination-desc': 'false',
+                },
+            };
+        }) as any);
+
+    test('continues past a page whose items were all filtered out', async () => {
+        // The first chunk scans 1000 rows and returns none of them; the items behind it must still be yielded.
+        const mockedClient = mockTransformedDataset(2000, (rows) => rows.filter(({ id }) => Number(id) >= 1000));
+
+        try {
+            const items = [];
+            for await (const item of client.dataset('some-id').listItems({ clean: true, chunkSize: 1000 })) {
+                items.push(item);
+            }
+            expect(items).toEqual(range(1000, 2000));
+            expect(mockedClient.mock.calls.map(([request]: any) => request.params.offset)).toEqual([undefined, 1000]);
+        } finally {
+            mockedClient.mockRestore();
+        }
+    });
+
+    test('advances the offset by the rows scanned rather than the items returned', async () => {
+        // Every chunk of two rows returns one. Advancing by `items.length` would make the next page re-scan, and
+        // re-yield, a row the previous page already returned.
+        const mockedClient = mockTransformedDataset(4, (rows) => rows.filter(({ id }) => Number(id) % 2 === 0));
+
+        try {
+            const items = [];
+            for await (const item of client.dataset('some-id').listItems({ clean: true, chunkSize: 2 })) {
+                items.push(item);
+            }
+            expect(items).toEqual([...range(0, 1), ...range(2, 3)]);
+            expect(mockedClient.mock.calls.map(([request]: any) => request.params.offset)).toEqual([undefined, 2]);
+        } finally {
+            mockedClient.mockRestore();
+        }
+    });
+
+    test('advances the offset by the rows scanned when unwind returns more items than rows', async () => {
+        // Every row unwinds into three items. Advancing by `items.length` would push the offset past rows the next
+        // page has not scanned yet, and skip them.
+        const unwind = (rows: ReturnType<typeof range>) =>
+            rows.flatMap((row) => ['a', 'b', 'c'].map((part) => ({ ...row, part })));
+        const mockedClient = mockTransformedDataset(4, unwind);
+
+        try {
+            const items = [];
+            for await (const item of client.dataset('some-id').listItems({ unwind: 'parts', chunkSize: 2 })) {
+                items.push(item);
+            }
+            expect(items).toEqual(unwind(range(0, 4)));
+            expect(mockedClient.mock.calls.map(([request]: any) => request.params.offset)).toEqual([undefined, 2]);
+        } finally {
+            mockedClient.mockRestore();
+        }
+    });
 });
 
 describe('KeyValueStoreClient.listKeys as async iterable', () => {