diff --git a/docs/chains/stellar-streaming-scan-pipeline.md b/docs/chains/stellar-streaming-scan-pipeline.md index 8392c31..31e13ff 100644 --- a/docs/chains/stellar-streaming-scan-pipeline.md +++ b/docs/chains/stellar-streaming-scan-pipeline.md @@ -80,3 +80,29 @@ mechanism directly (a mock producer/consumer pair with matched I/O and CPU delay asserts the bounded-queue backpressure property with a fast producer paired with an artificially slow consumer, so both acceptance criteria run under `pnpm test`, not just the excluded `bench/` folder. + +## Bounding parallel cold scans + +`fetchAnnouncementsStream` can split a cold scan's ledger range into parallel +chunks (`parallelism`), and `mergeOrdered` interleaves those chunk generators in +ledger order. Both halves are now bounded: + +- **Concurrent chunks.** `parallelism` is a caller hint, clamped to + `[1, MAX_COLD_SCAN_PARALLELISM]` (`MAX_COLD_SCAN_PARALLELISM = 8`). Each chunk + keeps one `getEvents` page in flight, so without the cap a large hint would + start an unbounded number of concurrent RPC requests for no throughput gain on + a single I/O-bound endpoint. +- **Buffered announcements.** `mergeOrdered` holds exactly one item per chunk + (`pending`) and only pulls from a chunk again after the consumer has read its + previous item. Memory is O(number of chunks), not O(scan size), so a slow + consumer cannot make a fast producer run ahead. +- **Cancellation.** `mergeOrdered` closes **every** chunk iterator in a `finally` + block, so breaking out of the consumer's `for-await` (or `.return()`) reaches + the chunk that was in flight and the ones that had not yielded yet. Previously + only the delegated iterator chain unwound, leaving other chunk generators + suspended with a page requested. + +`test/chains/stellar/announcements.test.ts` covers all three: the clamp on an +oversized `parallelism`, the per-chunk lead bound under a slow consumer, and +cancellation closing every chunk iterator. `test/chains/stellar/bench/scan.bench.ts` +adds a bounded-merge case under "Stellar cold-scan backpressure". diff --git a/etc/sdk-stellar.api.md b/etc/sdk-stellar.api.md index 0beb8d7..1ffe406 100644 --- a/etc/sdk-stellar.api.md +++ b/etc/sdk-stellar.api.md @@ -368,6 +368,7 @@ export interface FetchAnnouncementsOptions { includeV1?: boolean; includeV2?: boolean; parallelism?: number; + signal?: AbortSignal; sorobanUrl?: string; toLedger?: number; toTimestamp?: Date; diff --git a/src/chains/stellar/announcements.ts b/src/chains/stellar/announcements.ts index 155bc24..85f62a0 100644 --- a/src/chains/stellar/announcements.ts +++ b/src/chains/stellar/announcements.ts @@ -35,9 +35,28 @@ export interface FetchAnnouncementsOptions { /** * Number of parallel chunks to fetch for cold scans. Splits the ledger range * into N contiguous chunks fetched concurrently, then merged in order. - * Default: 1 (sequential). Ignored when cursor is provided. + * Default: 1 (sequential). Ignored when cursor is provided. Clamped to + * `[1, MAX_COLD_SCAN_PARALLELISM]`, so the caller cannot start an unbounded + * number of in-flight pages. */ parallelism?: number; + /** + * Standard `AbortSignal` used to cancel the scan. When it fires, every + * in-flight Soroban RPC and Horizon request is aborted, the parallel chunk + * iterators are closed, and the stream throws an `AbortError` (`DOMException`). + */ + signal?: AbortSignal; +} + +/** + * Throws a standard `AbortError` when the signal has already been aborted. + * In-flight `fetch` calls reject with the same error once the signal fires. + * @internal + */ +function throwIfAborted(signal?: AbortSignal): void { + if (signal?.aborted) { + throw new DOMException('This operation was aborted', 'AbortError'); + } } export class RetentionExceededError extends Error { @@ -209,14 +228,15 @@ async function* fetchAnnouncementsRange( startLedger: number, toLedger: number | undefined, seen: Set, + signal?: AbortSignal, ): AsyncGenerator<{ announcement: Announcement; ledger: number }> { - const singleFilterGroup = filterGroups.length === 1; - for (const filters of filterGroups) { let hasMore = true; let groupCursor: string | undefined = undefined; while (hasMore) { + throwIfAborted(signal); + const params: Record = { filters, pagination: groupCursor ? { limit: 1000, cursor: groupCursor } : { limit: 1000 }, @@ -235,6 +255,7 @@ async function* fetchAnnouncementsRange( method: 'getEvents', params, }), + signal, }); const data = assertRpcEnvelope(await res.json(), sorobanUrl); @@ -251,6 +272,8 @@ async function* fetchAnnouncementsRange( const events = assertEventsResult(data, sorobanUrl) ?? []; for (const event of events) { + throwIfAborted(signal); + assertEvent(event, sorobanUrl); const ledger = eventLedger(event); if (toLedger !== undefined && ledger !== undefined && ledger >= toLedger) { @@ -264,6 +287,7 @@ async function* fetchAnnouncementsRange( const ann = parseAnnouncementEvent(event, { endpoint: sorobanUrl }); if (ann && ledger !== undefined) { + throwIfAborted(signal); yield { announcement: ann, ledger }; } } @@ -284,36 +308,61 @@ async function* fetchAnnouncementsRange( */ export async function* mergeOrdered( iterables: Array>, + signal?: AbortSignal, ): AsyncGenerator { const iterators = iterables.map((it) => it[Symbol.asyncIterator]()); const pending: Array<{ value: T; key: number; index: number }> = []; - // Initialize: pull first item from each iterator - for (let i = 0; i < iterators.length; i++) { - const result = await iterators[i].next(); - if (!result.done) { - pending.push({ value: result.value.item, key: result.value.key, index: i }); + try { + throwIfAborted(signal); + + // Initialize: pull the first item from every iterator concurrently so a + // slow chunk cannot delay the first page of the others. + const firstResults = await Promise.all(iterators.map((it) => it.next())); + for (let i = 0; i < firstResults.length; i++) { + const result = firstResults[i]; + if (!result.done) { + pending.push({ value: result.value.item, key: result.value.key, index: i }); + } } - } - while (pending.length > 0) { - // Find the item with the smallest key - pending.sort((a, b) => a.key - b.key || a.index - b.index); - const [next, ...rest] = pending; + while (pending.length > 0) { + throwIfAborted(signal); - yield next.value; + // Find the item with the smallest key + pending.sort((a, b) => a.key - b.key || a.index - b.index); + const [next, ...rest] = pending; - // Pull the next item from the iterator that just yielded - const result = await iterators[next.index].next(); - if (!result.done) { - rest.push({ value: result.value.item, key: result.value.key, index: next.index }); - } + yield next.value; + + // Pull the next item from the iterator that just yielded + const result = await iterators[next.index].next(); + if (!result.done) { + rest.push({ value: result.value.item, key: result.value.key, index: next.index }); + } - pending.length = 0; - pending.push(...rest); + pending.length = 0; + pending.push(...rest); + } + } finally { + // Close every chunk iterator (including ones that never yielded) so that a + // consumer cancellation or an aborted signal releases each chunk's + // outstanding page instead of leaving it suspended. + await Promise.allSettled(iterators.map((it) => it.return?.())); } } +/** + * Hard ceiling on parallel cold-scan chunks. + * + * `parallelism` is a caller hint, and every chunk keeps a `getEvents` page in + * flight and one buffered item per chunk in the ordered merge, so an + * unbounded value would mean unbounded pending RPC work and memory. Cold + * scans are I/O-bound on a single Soroban endpoint, so more than this many + * concurrent pages buys no throughput while multiplying load and buffer size. + */ +export const MAX_COLD_SCAN_PARALLELISM = 8; + /** * Splits a ledger range into N contiguous chunks. * @internal @@ -347,6 +396,9 @@ export function splitRange( * from the Soroban RPC as they arrive, never holding more than one page in memory. * * Cancellation is automatic: breaking out of the `for-await` loop stops the stream. + * Passing `options.signal` additionally aborts every in-flight Soroban RPC and + * Horizon request, closes the parallel chunk iterators, and throws a standard + * `AbortError` (`DOMException`). * * @param chain The chain identifier (default: "stellar"). * @param sorobanUrlOrOpts Optional override for the Soroban RPC URL, or FetchAnnouncementsOptions. @@ -359,6 +411,10 @@ export async function* fetchAnnouncementsStream( ): AsyncGenerator { const deployment = getDeployment(chain); const opts = typeof sorobanUrlOrOpts === 'object' ? sorobanUrlOrOpts : maybeOpts; + const signal = opts?.signal; + + throwIfAborted(signal); + const sorobanUrl = (typeof sorobanUrlOrOpts === 'string' ? sorobanUrlOrOpts : opts?.sorobanUrl) || deployment.sorobanUrl; @@ -374,25 +430,31 @@ export async function* fetchAnnouncementsStream( throw new Error('toLedger and toTimestamp are mutually exclusive'); } - const ledgerWindow = await getSorobanLedgerWindow(sorobanUrl, announcerContract); - const latestLedger = ledgerWindow.latest ?? (await getLatestLedger(sorobanUrl)); + const ledgerWindow = await getSorobanLedgerWindow(sorobanUrl, announcerContract, signal); + const latestLedger = ledgerWindow.latest ?? (await getLatestLedger(sorobanUrl, signal)); let startLedger = opts?.fromLedger ?? Math.max(ledgerWindow.oldest ?? 1, latestLedger ? latestLedger - 5000 : 1); let toLedger = opts?.toLedger ?? latestLedger; if (opts?.fromTimestamp) { - startLedger = await ledgerForTimestamp(deployment.horizonUrl, opts.fromTimestamp); + startLedger = await ledgerForTimestamp(deployment.horizonUrl, opts.fromTimestamp, signal); } if (opts?.toTimestamp) { - toLedger = await ledgerForTimestamp(deployment.horizonUrl, opts.toTimestamp); + toLedger = await ledgerForTimestamp(deployment.horizonUrl, opts.toTimestamp, signal); } if (!opts?.cursor && ledgerWindow.oldest !== undefined && startLedger < ledgerWindow.oldest) { throw new RetentionExceededError(startLedger, ledgerWindow.oldest); } - // Use parallel chunking for cold scans (no cursor) when parallelism > 1 - const parallelism = opts?.parallelism ?? 1; + // Use parallel chunking for cold scans (no cursor) when parallelism > 1. + // The requested value is clamped: each chunk keeps one page in flight and + // one item buffered in the merge, so an unbounded hint would mean unbounded + // pending work. Non-finite / fractional hints fall back to 1. + const requestedParallelism = Number.isFinite(opts?.parallelism) + ? Math.floor(opts?.parallelism as number) + : 1; + const parallelism = Math.min(MAX_COLD_SCAN_PARALLELISM, Math.max(1, requestedParallelism)); if (!opts?.cursor && parallelism > 1 && toLedger !== undefined) { const seen = new Set(); const chunks = splitRange(startLedger, toLedger, parallelism); @@ -406,13 +468,14 @@ export async function* fetchAnnouncementsStream( chunk.startLedger, chunk.endLedger, seen, + signal, )) { yield { item: result.announcement, key: result.ledger }; } })(); }); - yield* mergeOrdered(chunkIterables); + yield* mergeOrdered(chunkIterables, signal); return; } @@ -426,6 +489,8 @@ export async function* fetchAnnouncementsStream( let groupCursor = singleFilterGroup ? cursor : undefined; while (hasMore) { + throwIfAborted(signal); + const params: Record = { filters, pagination: groupCursor ? { limit: 1000, cursor: groupCursor } : { limit: 1000 }, @@ -444,6 +509,7 @@ export async function* fetchAnnouncementsStream( method: 'getEvents', params, }), + signal, }); const data = assertRpcEnvelope(await res.json(), sorobanUrl); @@ -461,6 +527,8 @@ export async function* fetchAnnouncementsStream( const events = assertEventsResult(data, sorobanUrl) ?? []; for (const event of events) { + throwIfAborted(signal); + assertEvent(event, sorobanUrl); const ledger = eventLedger(event); if (toLedger !== undefined && ledger !== undefined && ledger >= toLedger) { @@ -473,7 +541,10 @@ export async function* fetchAnnouncementsStream( seen.add(dedupeKey); const ann = parseAnnouncementEvent(event, { endpoint: sorobanUrl }); - if (ann) yield ann; + if (ann) { + throwIfAborted(signal); + yield ann; + } } if (!hasMore || events.length < 1000) { @@ -489,7 +560,10 @@ export async function* fetchAnnouncementsStream( async function getSorobanLedgerWindow( sorobanUrl: string, announcerContract: string, + signal?: AbortSignal, ): Promise<{ oldest?: number; latest?: number }> { + throwIfAborted(signal); + const probeRes = await fetch(sorobanUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, @@ -503,6 +577,7 @@ async function getSorobanLedgerWindow( pagination: { limit: 1 }, }, }), + signal, }); const probeData = assertRpcEnvelope(await probeRes.json(), sorobanUrl); @@ -513,11 +588,17 @@ async function getSorobanLedgerWindow( return {}; } -async function getLatestLedger(sorobanUrl: string): Promise { +async function getLatestLedger( + sorobanUrl: string, + signal?: AbortSignal, +): Promise { + throwIfAborted(signal); + const res = await fetch(sorobanUrl, { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ jsonrpc: '2.0', id: 1, method: 'getLatestLedger' }), + signal, }); const data = assertRpcEnvelope(await res.json(), sorobanUrl); if (data.error !== undefined) return undefined; @@ -534,15 +615,21 @@ async function getLatestLedger(sorobanUrl: string): Promise return sequence as number; } -async function ledgerForTimestamp(horizonUrl: string, timestamp: Date): Promise { - const latest = await horizonLedger(horizonUrl, 'latest'); +async function ledgerForTimestamp( + horizonUrl: string, + timestamp: Date, + signal?: AbortSignal, +): Promise { + const latest = await horizonLedger(horizonUrl, 'latest', signal); let low = 1; let high = latest.sequence; let answer = latest.sequence + 1; while (low <= high) { + throwIfAborted(signal); + const mid = Math.floor((low + high) / 2); - const ledger = await horizonLedger(horizonUrl, mid); + const ledger = await horizonLedger(horizonUrl, mid, signal); const closedAt = Date.parse(ledger.closed_at); if (closedAt >= timestamp.getTime()) { @@ -559,12 +646,15 @@ async function ledgerForTimestamp(horizonUrl: string, timestamp: Date): Promise< async function horizonLedger( horizonUrl: string, sequence: number | 'latest', + signal?: AbortSignal, ): Promise<{ sequence: number; closed_at: string }> { + throwIfAborted(signal); + const path = sequence === 'latest' ? '/ledgers?order=desc&limit=1' : `/ledgers/${encodeURIComponent(sequence)}`; - const res = await fetch(`${horizonUrl}${path}`); + const res = await fetch(`${horizonUrl}${path}`, { signal }); const data = await res.json(); if (sequence === 'latest') { return data._embedded.records[0]; diff --git a/test/chains/stellar/announcements.test.ts b/test/chains/stellar/announcements.test.ts index bcb4249..72112d1 100644 --- a/test/chains/stellar/announcements.test.ts +++ b/test/chains/stellar/announcements.test.ts @@ -571,6 +571,212 @@ describe('parallel chunking ordering guarantee', () => { expect(scan.startLedger).toBeUndefined(); expect(scan.pagination).toEqual({ limit: 1000, cursor: 'previous-cursor' }); }); + + test('caps cold-scan parallelism so in-flight chunks stay bounded', async () => { + const chunkStarts: number[] = []; + let call = 0; + fetchSpy = vi.fn(async (_input: RequestInfo | URL, init?: RequestInit) => { + call += 1; + const body = init?.body ? JSON.parse(init.body.toString()) : undefined; + // 1: ledger-window probe, 2: latest-ledger lookup, 3+: chunk pages. + if (call === 1) return { json: async () => rpcEnvelope(makeProbeSuccess(), 0) } as Response; + if (call === 2) + return { json: async () => rpcEnvelope({ result: { sequence: 100 } }, 1) } as Response; + chunkStarts.push(body?.params?.startLedger); + return { + json: async () => rpcEnvelope({ result: { events: [], cursor: undefined } }, 2), + } as Response; + }); + vi.stubGlobal('fetch', fetchSpy); + + const { MAX_COLD_SCAN_PARALLELISM } = await import('../../../src/chains/stellar/announcements'); + + await collectStream( + fetchAnnouncementsStream('stellar', { + fromLedger: 1_000, + toLedger: 100_000, + parallelism: 10_000, // far above the cap + includeV2: false, + }), + ); + + // An unbounded hint must not fan out into unbounded chunk generators. + expect(chunkStarts).toHaveLength(MAX_COLD_SCAN_PARALLELISM); + expect(new Set(chunkStarts).size).toBe(MAX_COLD_SCAN_PARALLELISM); + expect(MAX_COLD_SCAN_PARALLELISM).toBeGreaterThan(1); + }); + + test('mergeOrdered closes every chunk iterator when the consumer cancels', async () => { + const { mergeOrdered } = await import('../../../src/chains/stellar/announcements'); + + const closed: number[] = []; + const iterables = Array.from({ length: 5 }, (_, index) => + (async function* () { + try { + for (let i = 0; i < 100; i++) { + yield { item: index * 100 + i, key: index * 100 + i }; + } + } finally { + closed.push(index); + } + })(), + ); + + const seen: number[] = []; + for await (const item of mergeOrdered(iterables)) { + seen.push(item); + if (seen.length === 1) break; + } + + expect(seen).toEqual([0]); + // Every chunk must be returned, not just the one that was in flight. + expect([...closed].sort((a, b) => a - b)).toEqual([0, 1, 2, 3, 4]); + }); + + test('mergeOrdered bounds how far each chunk runs ahead of a slow consumer', async () => { + const { mergeOrdered } = await import('../../../src/chains/stellar/announcements'); + + const chunkCount = 4; + const produced = new Array(chunkCount).fill(0); + const iterables = produced.map((_value, index) => + (async function* () { + for (let i = 0; i < 100; i++) { + produced[index] += 1; + // Interleaved keys so the merge round-robins instead of draining one chunk. + yield { item: index, key: i * chunkCount + index }; + } + })(), + ); + + let consumed = 0; + for await (const _item of mergeOrdered(iterables)) { + consumed += 1; + await sleep(1); // deliberately slow consumer + if (consumed === 20) break; + } + + // One item buffered per chunk plus the one in flight: no unbounded buffering. + expect(Math.max(...produced)).toBeLessThanOrEqual(Math.ceil(consumed / chunkCount) + 2); + }); +}); + +// --------------------------------------------------------------------------- +// AbortSignal cancellation and concurrent initial fetch +// --------------------------------------------------------------------------- + +describe('AbortSignal cancellation', () => { + test('pre-aborted signal throws AbortError before any RPC request', async () => { + const { fetchAnnouncementsStream } = await import('../../../src/chains/stellar/announcements'); + const spy = vi.fn(); + vi.stubGlobal('fetch', spy); + + const controller = new AbortController(); + controller.abort(); + + const stream = fetchAnnouncementsStream('stellar', { signal: controller.signal }); + await expect(stream.next()).rejects.toMatchObject({ name: 'AbortError' }); + expect(spy).not.toHaveBeenCalled(); + }); + + test('threads the signal into fetch and cancels an in-flight page request', async () => { + const { fetchAnnouncementsStream } = await import('../../../src/chains/stellar/announcements'); + const controller = new AbortController(); + + const blockingFetch = vi.fn((_input: RequestInfo | URL, init?: RequestInit) => { + const body = init?.body ? JSON.parse(String(init.body)) : {}; + if (body.method === 'getEvents' && body.id === 0) { + return Promise.resolve({ + json: async () => rpcEnvelope(makeProbeSuccess(), 0), + } as Response); + } + if (body.method === 'getLatestLedger') { + return Promise.resolve({ + json: async () => rpcEnvelope({ result: { sequence: 100 } }, 1), + } as Response); + } + // Simulate a page request that stays in flight until it is aborted. + const signal = init?.signal; + return new Promise((_resolve, reject) => { + const onAbort = () => reject(new DOMException('Aborted', 'AbortError')); + if (signal?.aborted) { + onAbort(); + return; + } + signal?.addEventListener('abort', onAbort, { once: true }); + setTimeout(() => controller.abort(), 0); + }); + }); + vi.stubGlobal('fetch', blockingFetch); + + await expect( + collectStream(fetchAnnouncementsStream('stellar', { signal: controller.signal })), + ).rejects.toMatchObject({ name: 'AbortError' }); + + const pageCall = blockingFetch.mock.calls.find( + ([, init]) => JSON.parse(String(init?.body)).id === 2, + ); + expect(pageCall).toBeDefined(); + expect(pageCall?.[1]?.signal).toBe(controller.signal); + }); + + test('mergeOrdered pulls the first item from every iterator concurrently', async () => { + const { mergeOrdered } = await import('../../../src/chains/stellar/announcements'); + + const started: number[] = []; + const gates: Array<() => void> = []; + const makeChunk = (index: number) => + (async function* () { + started.push(index); + await new Promise((resolve) => gates.push(resolve)); + yield { item: index, key: index }; + })(); + + const iterator = mergeOrdered([makeChunk(0), makeChunk(1), makeChunk(2)])[ + Symbol.asyncIterator + ](); + const first = iterator.next(); + + // Sequential initialisation would only have started the first chunk here. + expect([...started].sort((a, b) => a - b)).toEqual([0, 1, 2]); + + gates.forEach((release) => release()); + await expect(first).resolves.toMatchObject({ done: false, value: 0 }); + await iterator.return?.(undefined); + }); + + test('mergeOrdered closes every chunk iterator when the signal aborts', async () => { + const { mergeOrdered } = await import('../../../src/chains/stellar/announcements'); + const controller = new AbortController(); + let closed = 0; + + const makeChunk = (start: number) => + (async function* () { + try { + for (let i = 0; ; i++) { + await sleep(1); + yield { item: start + i, key: start + i }; + } + } finally { + closed++; + } + })(); + + const seen: number[] = []; + await expect( + (async () => { + for await (const item of mergeOrdered( + [makeChunk(1), makeChunk(10), makeChunk(20)], + controller.signal, + )) { + seen.push(item); + if (seen.length === 2) controller.abort(); + } + })(), + ).rejects.toMatchObject({ name: 'AbortError' }); + + expect(seen).toHaveLength(2); + expect(closed).toBe(3); + }); }); function sleep(ms: number): Promise { diff --git a/test/chains/stellar/bench/scan.bench.ts b/test/chains/stellar/bench/scan.bench.ts index edf3ca1..2f5cc15 100644 --- a/test/chains/stellar/bench/scan.bench.ts +++ b/test/chains/stellar/bench/scan.bench.ts @@ -15,7 +15,11 @@ import { import { SCHEME_ID } from '../../../../src/chains/stellar/constants'; import { bytesToHex } from '../../../../src/chains/stellar/utils'; import type { Announcement, StealthKeys } from '../../../../src/chains/stellar/types'; -import { fetchAnnouncementsStream } from '../../../../src/chains/stellar/announcements'; +import { + fetchAnnouncementsStream, + mergeOrdered, + MAX_COLD_SCAN_PARALLELISM, +} from '../../../../src/chains/stellar/announcements'; const MATCH_INDEX = 997; const POOL_SIZE = 512; @@ -345,3 +349,54 @@ describe('Stellar parallel horizon range chunking', () => { BENCH_OPTIONS, ); }); + +/** + * Backpressure for a cold scan's parallel chunks. `mergeOrdered` buffers one + * item per chunk and only pulls again after the consumer reads, so the cap on + * chunk count is what keeps a slow consumer's pending work bounded. + */ +describe('Stellar cold-scan backpressure', () => { + const CHUNK_COUNT = MAX_COLD_SCAN_PARALLELISM; + const ITEMS_PER_CHUNK = 2_000; + const PAGE_LATENCY_MS = 10; + + test('bounded merge over the maximum chunk count drains in order', async () => { + const iterables = Array.from({ length: CHUNK_COUNT }, (_, index) => + (async function* () { + await sleep(PAGE_LATENCY_MS); + for (let i = 0; i < ITEMS_PER_CHUNK; i++) { + yield { item: index, key: i * CHUNK_COUNT + index }; + } + })(), + ); + + let consumed = 0; + for await (const _item of mergeOrdered(iterables)) { + consumed += 1; + if (consumed % 500 === 0) await sleep(1); + } + expect(consumed).toBe(CHUNK_COUNT * ITEMS_PER_CHUNK); + }); + + bench( + `bounded merge buffer, slow consumer (${CHUNK_COUNT} chunks)`, + async () => { + const iterables = Array.from({ length: CHUNK_COUNT }, (_, index) => + (async function* () { + await sleep(PAGE_LATENCY_MS); + for (let i = 0; i < ITEMS_PER_CHUNK; i++) { + yield { item: index, key: i * CHUNK_COUNT + index }; + } + })(), + ); + + let consumed = 0; + for await (const _item of mergeOrdered(iterables)) { + consumed += 1; + if (consumed % 500 === 0) await sleep(1); + } + expect(consumed).toBe(CHUNK_COUNT * ITEMS_PER_CHUNK); + }, + BENCH_OPTIONS, + ); +});