From a503b9356729c68750c5c79ef2029a95b577075a Mon Sep 17 00:00:00 2001 From: Stephan-Thomas Date: Tue, 23 Jun 2026 12:19:50 +0100 Subject: [PATCH] Implement adaptive polling for streams --- src/account/streamAccount.ts | 74 ++++++++++++- src/tests/account.test.ts | 110 ++++++++++++++++++- src/tests/transaction.test.ts | 148 ++++++++++++++++++++++++++ src/transaction/streamTransactions.ts | 76 ++++++++++++- 4 files changed, 403 insertions(+), 5 deletions(-) create mode 100644 src/tests/transaction.test.ts diff --git a/src/account/streamAccount.ts b/src/account/streamAccount.ts index c7226d6..4bb5104 100644 --- a/src/account/streamAccount.ts +++ b/src/account/streamAccount.ts @@ -4,6 +4,10 @@ import { sleep, toMessage } from "../shared"; import type { AccountInfo } from "./types"; import { getAccount } from "./getAccount"; +const MIN_POLL_INTERVAL_MS = 1_000; +const DEFAULT_POLL_INTERVAL_MS = 5_000; +const ADAPTIVE_INTERVAL_STEP_MS = 1_000; + /** * Configuration for account streaming. */ @@ -13,6 +17,21 @@ export interface AccountStreamConfig { * Minimum enforced: 1000 ms to avoid hammering Horizon. */ intervalMs?: number; + /** + * Minimum polling interval in milliseconds when adaptive polling is enabled. + * Default: 1000 ms. + */ + minIntervalMs?: number; + /** + * Maximum polling interval in milliseconds when adaptive polling is enabled. + * Default: the base interval. + */ + maxIntervalMs?: number; + /** + * Number of unchanged polls before increasing the interval. + * Default: 3. + */ + adaptiveThreshold?: number; /** * Maximum number of polls before the stream ends. * Omit for an infinite stream. @@ -49,11 +68,55 @@ export async function* streamAccount( config?: AccountStreamConfig, signal?: AbortSignal, ): AsyncGenerator> { - const intervalMs = Math.max(config?.intervalMs ?? 5_000, 1_000); + const baseIntervalMs = Math.max( + config?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS, + MIN_POLL_INTERVAL_MS, + ); + const adaptiveEnabled = + config?.minIntervalMs !== undefined || + config?.maxIntervalMs !== undefined || + config?.adaptiveThreshold !== undefined; + const minIntervalMs = Math.max( + config?.minIntervalMs ?? MIN_POLL_INTERVAL_MS, + MIN_POLL_INTERVAL_MS, + ); + const maxIntervalMs = Math.max( + config?.maxIntervalMs ?? baseIntervalMs, + minIntervalMs, + ); + const adaptiveThreshold = Math.max(config?.adaptiveThreshold ?? 3, 1); const maxPolls = config?.maxPolls; const emitOnStart = config?.emitOnStart ?? true; let polls = 0; + let currentIntervalMs = Math.min( + Math.max(baseIntervalMs, minIntervalMs), + maxIntervalMs, + ); + let unchangedPolls = 0; + let lastSnapshot: string | null = null; + + const adjustInterval = (changed: boolean): void => { + if (!adaptiveEnabled) return; + + if (changed) { + unchangedPolls = 0; + currentIntervalMs = Math.max( + minIntervalMs, + currentIntervalMs - ADAPTIVE_INTERVAL_STEP_MS, + ); + return; + } + + unchangedPolls++; + if (unchangedPolls < adaptiveThreshold) return; + + unchangedPolls = 0; + currentIntervalMs = Math.min( + maxIntervalMs, + currentIntervalMs + ADAPTIVE_INTERVAL_STEP_MS, + ); + }; while (true) { if (signal?.aborted) return; @@ -64,7 +127,7 @@ export async function* streamAccount( // Skip the initial sleep when emitOnStart is true if (polls > 0 || !emitOnStart) { try { - await sleep(intervalMs); + await sleep(currentIntervalMs); } catch { return; } @@ -74,8 +137,15 @@ export async function* streamAccount( try { const result = await getAccount(horizonUrl, publicKey); + const snapshot = + result.status === "ok" ? JSON.stringify(result.data) : null; + if (lastSnapshot !== null) { + adjustInterval(snapshot !== lastSnapshot); + } + lastSnapshot = snapshot; yield result; } catch (cause) { + adjustInterval(false); yield err( SorokitErrorCode.ACCOUNT_FETCH_FAILED, `Account stream poll failed: ${toMessage(cause)}`, diff --git a/src/tests/account.test.ts b/src/tests/account.test.ts index 91a6bae..a3cae06 100644 --- a/src/tests/account.test.ts +++ b/src/tests/account.test.ts @@ -1,5 +1,52 @@ -import { describe, it, expect } from "vitest"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import { formatAddress } from "../shared/utils"; +import { ok } from "../shared/response"; +import type { AccountInfo } from "../account/types"; + +const accountMockState = vi.hoisted(() => ({ + sleepCalls: [] as number[], + results: [] as AccountInfo[], + index: 0, +})); + +vi.mock("../shared", async () => { + const actual = await vi.importActual("../shared"); + return { + ...actual, + sleep: vi.fn((ms: number) => { + accountMockState.sleepCalls.push(ms); + return Promise.resolve(); + }), + }; +}); + +vi.mock("../account/getAccount", () => ({ + getAccount: vi.fn(async () => { + const result = + accountMockState.results[accountMockState.index] ?? + accountMockState.results.at(-1)!; + accountMockState.index++; + return ok(result); + }), +})); + +import { streamAccount } from "../account/streamAccount"; + +function createAccount(sequence: string): AccountInfo { + return { + publicKey: "GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA", + displayAddress: "GAAAA...AAAA", + sequence, + subentryCount: 0, + balances: [], + }; +} + +beforeEach(() => { + accountMockState.sleepCalls.length = 0; + accountMockState.index = 0; + accountMockState.results = []; +}); describe("account", () => { describe("formatAddress (pure utility — returns string, not SorokitResult)", () => { @@ -12,4 +59,65 @@ describe("account", () => { expect(formatAddress("GABCD")).toBe("GABCD"); }); }); + + describe("streamAccount", () => { + it("increases interval after unchanged polls and decreases after activity", async () => { + accountMockState.results = [ + createAccount("1"), + createAccount("1"), + createAccount("1"), + createAccount("2"), + ]; + + const stream = streamAccount("https://horizon.test", "G...", { + intervalMs: 2000, + minIntervalMs: 1000, + maxIntervalMs: 4000, + adaptiveThreshold: 2, + maxPolls: 4, + }); + + await stream.next(); + await stream.next(); + await stream.next(); + await stream.next(); + + expect(accountMockState.sleepCalls).toEqual([2000, 2000, 3000]); + }); + + it("respects interval boundaries", async () => { + accountMockState.results = [ + createAccount("1"), + createAccount("1"), + createAccount("1"), + createAccount("1"), + createAccount("2"), + createAccount("2"), + createAccount("2"), + createAccount("2"), + ]; + + const stream = streamAccount("https://horizon.test", "G...", { + intervalMs: 2000, + minIntervalMs: 1000, + maxIntervalMs: 3000, + adaptiveThreshold: 1, + maxPolls: 8, + }); + + for (let i = 0; i < 8; i++) { + await stream.next(); + } + + expect(accountMockState.sleepCalls).toEqual([ + 2000, + 3000, + 3000, + 3000, + 2000, + 1000, + 1000, + ]); + }); + }); }); diff --git a/src/tests/transaction.test.ts b/src/tests/transaction.test.ts new file mode 100644 index 0000000..7b1925f --- /dev/null +++ b/src/tests/transaction.test.ts @@ -0,0 +1,148 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; +import type { TransactionPage } from "../transaction/streamTransactions"; + +const transactionMockState = vi.hoisted(() => ({ + sleepCalls: [] as number[], + pages: [] as TransactionPage[], + index: 0, +})); + +vi.mock("../shared", async () => { + const actual = await vi.importActual("../shared"); + return { + ...actual, + sleep: vi.fn((ms: number) => { + transactionMockState.sleepCalls.push(ms); + return Promise.resolve(); + }), + }; +}); + +vi.mock("@stellar/stellar-sdk", () => { + class Server { + constructor() {} + + transactions() { + const buildCall = async () => { + const page = + transactionMockState.pages[transactionMockState.index] ?? + transactionMockState.pages.at(-1)!; + transactionMockState.index++; + return { + records: page.transactions.map((tx, index) => ({ + hash: tx.hash, + successful: tx.status === "success", + ledger_attr: tx.ledger, + created_at: tx.createdAt, + fee_charged: tx.fee, + envelope_xdr: tx.envelopeXdr ?? "", + result_xdr: tx.resultXdr ?? "", + paging_token: + index === page.transactions.length - 1 + ? page.nextCursor + : `${page.nextCursor ?? "0"}-${index}`, + })), + }; + }; + + const builder = { + call: buildCall, + cursor: () => builder, + }; + + return { + forAccount: () => ({ + limit: () => ({ + order: () => builder, + }), + }), + }; + } + } + + return { Horizon: { Server } }; +}); + +import { streamTransactions } from "../transaction/streamTransactions"; + +function createPage(nextCursor: string | null, hashSuffix: string): TransactionPage { + return { + transactions: [ + { + hash: `hash-${hashSuffix}`, + status: "success", + ledger: 1000, + createdAt: "2024-01-01T00:00:00Z", + fee: "100", + }, + ], + nextCursor, + }; +} + +beforeEach(() => { + transactionMockState.sleepCalls.length = 0; + transactionMockState.index = 0; + transactionMockState.pages = []; +}); + +describe("streamTransactions", () => { + it("increases interval after unchanged polls and decreases after activity", async () => { + transactionMockState.pages = [ + createPage("1", "a"), + createPage("1", "a"), + createPage("1", "a"), + createPage("2", "b"), + ]; + + const stream = streamTransactions("https://horizon.test", "G...", { + intervalMs: 2000, + minIntervalMs: 1000, + maxIntervalMs: 4000, + adaptiveThreshold: 2, + maxPolls: 4, + }); + + await stream.next(); + await stream.next(); + await stream.next(); + await stream.next(); + + expect(transactionMockState.sleepCalls).toEqual([2000, 2000, 3000]); + }); + + it("respects interval boundaries", async () => { + transactionMockState.pages = [ + createPage("1", "a"), + createPage("1", "a"), + createPage("1", "a"), + createPage("1", "a"), + createPage("2", "b"), + createPage("2", "b"), + createPage("2", "b"), + createPage("2", "b"), + ]; + + const stream = streamTransactions("https://horizon.test", "G...", { + intervalMs: 2000, + minIntervalMs: 1000, + maxIntervalMs: 3000, + adaptiveThreshold: 1, + maxPolls: 8, + }); + + for (let i = 0; i < 8; i++) { + await stream.next(); + } + + expect(transactionMockState.sleepCalls).toEqual([ + 2000, + 3000, + 3000, + 3000, + 2000, + 1000, + 1000, + ]); + }); +}); diff --git a/src/transaction/streamTransactions.ts b/src/transaction/streamTransactions.ts index 1ae4830..139aa72 100644 --- a/src/transaction/streamTransactions.ts +++ b/src/transaction/streamTransactions.ts @@ -4,6 +4,10 @@ import type { SorokitResult } from "../shared/response"; import { sleep, toMessage, isNotFoundError } from "../shared"; import type { TransactionResult } from "./types"; +const MIN_POLL_INTERVAL_MS = 1_000; +const DEFAULT_POLL_INTERVAL_MS = 5_000; +const ADAPTIVE_INTERVAL_STEP_MS = 1_000; + /** * Configuration for transaction streaming. */ @@ -13,6 +17,21 @@ export interface TransactionStreamConfig { * Minimum enforced: 1000 ms. */ intervalMs?: number; + /** + * Minimum polling interval in milliseconds when adaptive polling is enabled. + * Default: 1000 ms. + */ + minIntervalMs?: number; + /** + * Maximum polling interval in milliseconds when adaptive polling is enabled. + * Default: the base interval. + */ + maxIntervalMs?: number; + /** + * Number of unchanged polls before increasing the interval. + * Default: 3. + */ + adaptiveThreshold?: number; /** * Maximum number of polls before the stream ends. * Omit for an infinite stream. @@ -71,7 +90,23 @@ export async function* streamTransactions( config?: TransactionStreamConfig, signal?: AbortSignal, ): AsyncGenerator> { - const intervalMs = Math.max(config?.intervalMs ?? 5_000, 1_000); + const baseIntervalMs = Math.max( + config?.intervalMs ?? DEFAULT_POLL_INTERVAL_MS, + MIN_POLL_INTERVAL_MS, + ); + const adaptiveEnabled = + config?.minIntervalMs !== undefined || + config?.maxIntervalMs !== undefined || + config?.adaptiveThreshold !== undefined; + const minIntervalMs = Math.max( + config?.minIntervalMs ?? MIN_POLL_INTERVAL_MS, + MIN_POLL_INTERVAL_MS, + ); + const maxIntervalMs = Math.max( + config?.maxIntervalMs ?? baseIntervalMs, + minIntervalMs, + ); + const adaptiveThreshold = Math.max(config?.adaptiveThreshold ?? 3, 1); const maxPolls = config?.maxPolls; const limit = config?.limit ?? 10; const order = config?.order ?? "desc"; @@ -79,6 +114,34 @@ export async function* streamTransactions( let cursor = config?.cursor; let polls = 0; + let currentIntervalMs = Math.min( + Math.max(baseIntervalMs, minIntervalMs), + maxIntervalMs, + ); + let unchangedPolls = 0; + let lastSnapshot: string | null = null; + + const adjustInterval = (changed: boolean): void => { + if (!adaptiveEnabled) return; + + if (changed) { + unchangedPolls = 0; + currentIntervalMs = Math.max( + minIntervalMs, + currentIntervalMs - ADAPTIVE_INTERVAL_STEP_MS, + ); + return; + } + + unchangedPolls++; + if (unchangedPolls < adaptiveThreshold) return; + + unchangedPolls = 0; + currentIntervalMs = Math.min( + maxIntervalMs, + currentIntervalMs + ADAPTIVE_INTERVAL_STEP_MS, + ); + }; while (true) { if (signal?.aborted) return; @@ -87,7 +150,7 @@ export async function* streamTransactions( if (polls > 0 || !emitOnStart) { try { - await sleep(intervalMs); + await sleep(currentIntervalMs); } catch { return; } @@ -124,8 +187,17 @@ export async function* streamTransactions( const lastRecord = page.records[page.records.length - 1]; const nextCursor = lastRecord?.paging_token ?? null; + const snapshot = JSON.stringify({ + transactions, + nextCursor, + }); + if (lastSnapshot !== null) { + adjustInterval(snapshot !== lastSnapshot); + } + lastSnapshot = snapshot; yield ok({ transactions, nextCursor }); } catch (cause) { + adjustInterval(false); yield err( isNotFoundError(cause) ? SorokitErrorCode.ACCOUNT_NOT_FOUND