From b4d873c1c15015f6b8b64e8dbfb3990bfc34f3a7 Mon Sep 17 00:00:00 2001 From: markdavid000 Date: Mon, 31 Aug 2026 08:43:54 +0100 Subject: [PATCH 1/3] fix(nonce): clear dangling setTimeout in acquireWithFallback on success Capture the timer ID from setTimeout and call clearTimeout in a finally block after Promise.race resolves. This prevents the timer from keeping the Node event loop alive after a successful acquire, which could delay clean process exit and trigger open-handle warnings in test runs. Fixes #550 --- src/nonce/NonceManager.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) 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); } } From 35521a91188fc00c039aac3c2ce21464d0e18fc6 Mon Sep 17 00:00:00 2001 From: markdavid000 Date: Mon, 31 Aug 2026 08:44:01 +0100 Subject: [PATCH 2/3] fix(streams): cap concurrent RPCs in list() page fetching Add a bounded concurrency helper (mapWithConcurrency) that processes items with at most N in-flight async calls. Use it for both the address pre-warm fan-out and the stream info fetching in pageFromFilteredIds. Previously list() issued up to 100 concurrent simulateTransaction RPCs per page via bare Promise.all, which routinely tripped provider rate limits and caused 429s for other in-flight SDK calls sharing the endpoint. The new default concurrency of 8 keeps throughput high while staying well under typical rate limits. Fixes #549 --- src/streams.ts | 33 +++++++++++++++++++++++++++++++-- 1 file changed, 31 insertions(+), 2 deletions(-) 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 { From a6034994ade62aad01c428515b1a713cf81f7107 Mon Sep 17 00:00:00 2001 From: markdavid000 Date: Mon, 31 Aug 2026 08:44:13 +0100 Subject: [PATCH 3/3] fix(module49): dense output array and complete cache key in processStreamBatch Two fixes in Module49: 1. Sparse array (Issue #548): processStreamBatch now pushes results into a dense array instead of assigning to pre-allocated slots, avoiding undefined holes when input items are falsy. Callers iterating the result no longer crash with 'Cannot read properties of undefined'. 2. Stale cache key (Issue #547): The processSingleItem cache key now includes all fields read by withdrawableLocal (cancelled, pausedAt, ratePerSecond, startTime, endTime) in addition to the previously captured id, withdrawn, paused, and nowSec. This prevents stale cache hits when stream state changes within the same wall-clock second. Fixes #548 Fixes #547 --- src/module49.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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);