fix: bound concurrent pages and cancellation for large cold scans

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".

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;

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<string>,
+  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<string, unknown> = {
         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<T>(
   iterables: Array<AsyncIterable<{ item: T; key: number }>>,
+  signal?: AbortSignal,
 ): AsyncGenerator<T> {
   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<Announcement> {
   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<string>();
     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<string, unknown> = {
         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<number | undefined> {
+async function getLatestLedger(
+  sorobanUrl: string,
+  signal?: AbortSignal,
+): Promise<number | undefined> {
+  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<number | undefined>
   return sequence as number;
 }
 
-async function ledgerForTimestamp(horizonUrl: string, timestamp: Date): Promise<number> {
-  const latest = await horizonLedger(horizonUrl, 'latest');
+async function ledgerForTimestamp(
+  horizonUrl: string,
+  timestamp: Date,
+  signal?: AbortSignal,
+): Promise<number> {
+  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];