diff --git a/src/module49.ts b/src/module49.ts index 87f568a..2f7b9da 100644 --- a/src/module49.ts +++ b/src/module49.ts @@ -59,7 +59,7 @@ export class Module49 { */ public processStreamBatch(items: StreamBatchItem49[]): Module49Result[] { const startTime = performance.now(); - const results: Module49Result[] = new Array(items.length); + const results: Module49Result[] = []; for (let i = 0; i < items.length; i += this.batchChunkSize) { const chunkEnd = Math.min(i + this.batchChunkSize, items.length); @@ -67,7 +67,7 @@ export class Module49 { const item = items[j]; if (!item) continue; - results[j] = this.processSingleItem(item); + results.push(this.processSingleItem(item)); } } @@ -83,7 +83,7 @@ export class Module49 { */ public processSingleItem(item: StreamBatchItem49): Module49Result { const nowSec = item.timestamp ?? Math.floor(Date.now() / 1000); - const cacheKey = `${item.id}_${item.stream.withdrawn.toString()}_${item.stream.paused ? 1 : 0}_${nowSec}`; + const cacheKey = `${item.id}_${item.stream.withdrawn.toString()}_${item.stream.paused ? 1 : 0}_${item.stream.cancelled ? 1 : 0}_${item.stream.pausedAt}_${item.stream.ratePerSecond.toString()}_${item.stream.startTime}_${item.stream.endTime}_${nowSec}`; if (this.enableOptimization) { const cached = this.cache.get(cacheKey); diff --git a/src/nonce/NonceManager.ts b/src/nonce/NonceManager.ts index f027729..17f18e9 100644 --- a/src/nonce/NonceManager.ts +++ b/src/nonce/NonceManager.ts @@ -184,8 +184,9 @@ export class NonceManager { const { promise, cancel } = this.enqueue(); + let timer: ReturnType | undefined; const timeoutPromise = new Promise((_, reject) => { - setTimeout(() => { + timer = setTimeout(() => { cancel(); reject(new Error(`NonceManager: acquire timed out after ${timeoutMs}ms`)); }, timeoutMs); @@ -198,6 +199,8 @@ export class NonceManager { throw err; } throw new Error(String(err)); + } finally { + if (timer) clearTimeout(timer); } } diff --git a/src/streams.ts b/src/streams.ts index 26c725c..024b449 100644 --- a/src/streams.ts +++ b/src/streams.ts @@ -52,6 +52,33 @@ import { ConduitError, RateLimitError, InsufficientBalanceError, StreamErrorCode * Tracks which v1-deprecated methods have already warned this session, so * repeated calls (e.g. in a hot loop) do not spam the console. */ +/** Default concurrency limit for bounded page-fetching (Issue #549). */ +const DEFAULT_LIST_CONCURRENCY = 8; + +/** + * Runs `fn` over `items` with at most `concurrency` in-flight calls. + * Preserves result ordering to match a naive `Promise.all` fan-out. + */ +async function mapWithConcurrency( + items: T[], + concurrency: number, + fn: (item: T) => Promise, +): Promise { + const results = new Array(items.length); + let index = 0; + + async function worker() { + while (index < items.length) { + const i = index++; + results[i] = await fn(items[i]); + } + } + + const workers = Array.from({ length: Math.min(concurrency, items.length) }, () => worker()); + await Promise.all(workers); + return results; +} + const _warnedDeprecations = new Set(); /** @@ -613,8 +640,10 @@ export class StreamsModule { // call would serially resolve the address and then simulate — 2 serial // RPCs per stream. Pre-warming collapses the address lookups into a // single parallel fan-out before the info simulations begin. - await Promise.all(ids.map(id => this._resolveAddr(id))); - const streams = await Promise.all(ids.map(id => this.get(id))); + // Bounded concurrency (#549) avoids hammering the RPC endpoint with + // up to 100 simultaneous simulateTransaction requests. + await mapWithConcurrency(ids, DEFAULT_LIST_CONCURRENCY, (id) => this._resolveAddr(id)); + const streams = await mapWithConcurrency(ids, DEFAULT_LIST_CONCURRENCY, (id) => this.get(id)); const hasNextPage = hasNextPageOverride ?? ids.length === limit; const totalCount = BigInt(offset + ids.length); return {