Implement adaptive polling for streams

src/account/streamAccount.ts

@@ -5,6 +5,10 @@ import type { SorokitLogger } from "../shared/logger";
 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.
  */
@@ -14,6 +18,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.
@@ -51,11 +70,55 @@ export async function* streamAccount(
   signal?: AbortSignal,
   logger?: SorokitLogger,
 ): AsyncGenerator<SorokitResult<AccountInfo>> {
-  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,
+    );
+  };
   let lastEmitted: AccountInfo | undefined;
 
   logger?.debug("account.stream", {
@@ -83,7 +146,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;
       }

src/tests/account.test.ts

@@ -1,3 +1,52 @@
+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<typeof import("../shared")>("../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 = [];
+});
 import { describe, it, expect, vi } from "vitest";
 import { formatAddress, deepEqual } from "../shared/utils";
 
@@ -17,6 +66,65 @@ describe("account", () => {
     });
   });
 
+  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,
+      ]);
+    });
   describe("deepEqual", () => {
     it("returns true for identical plain objects", () => {
       expect(deepEqual({ a: 1, b: [2, 3] }, { a: 1, b: [2, 3] })).toBe(true);

src/transaction/streamTransactions.ts

@@ -5,6 +5,10 @@ import { sleep, toMessage, isNotFoundError, deepEqual } from "../shared";
 import type { SorokitLogger } from "../shared/logger";
 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.
  */
@@ -14,6 +18,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.
@@ -73,14 +92,58 @@ export async function* streamTransactions(
   signal?: AbortSignal,
   logger?: SorokitLogger,
 ): AsyncGenerator<SorokitResult<TransactionPage>> {
-  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";
   const emitOnStart = config?.emitOnStart ?? true;
 
   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,
+    );
+  };
   let lastEmitted: TransactionPage | undefined;
 
   logger?.debug("transaction.stream", {
@@ -107,7 +170,7 @@ export async function* streamTransactions(
 
     if (polls > 0 || !emitOnStart) {
       try {
-        await sleep(intervalMs);
+        await sleep(currentIntervalMs);
       } catch {
         return;
       }