Fix #682, #683, #684, #685: batch size cap, AMM invariant tests, idle connection recycling, one-shot scheduling

src/__tests__/ammCalculator.test.ts

@@ -0,0 +1,80 @@
+import { describe, it, expect } from "vitest";
+import { estimateSwapOutput } from "../ammCalculator.js";
+import { InsufficientLiquidityError } from "../errors.js";
+
+// ---------------------------------------------------------------------------
+// Constant-product invariant preservation (issue #683)
+// ---------------------------------------------------------------------------
+
+function makePool(reserveIn: string, reserveOut: string) {
+  return {
+    reserves: [
+      { asset: "XLM", amount: reserveIn },
+      { asset: "USDC", amount: reserveOut },
+    ],
+  };
+}
+
+describe("constant-product invariant preservation", () => {
+  it("preserves k = reserveA * reserveB after a simulated swap", () => {
+    const reserveIn = 1_000_000_000_000n;
+    const reserveOut = 500_000_000_000n;
+    const pool = makePool(reserveIn.toString(), reserveOut.toString());
+    const k = reserveIn * reserveOut;
+
+    const inputAmount = 10_000_000n;
+    const result = estimateSwapOutput(pool, inputAmount.toString(), "XLM");
+
+    const newReserveIn = reserveIn + inputAmount;
+    const newReserveOut = reserveOut - BigInt(result.outputAmount);
+    const newK = newReserveIn * newReserveOut;
+
+    // Integer-division rounding means newK can only ever be <= k, and the
+    // relative drift must stay within a tight tolerance (no fee/rounding bug
+    // can silently break the invariant).
+    expect(newK).toBeLessThanOrEqual(k);
+    const drift = k - newK;
+    const tolerance = k / 1_000_000n; // 0.0001% relative tolerance
+    expect(drift).toBeLessThanOrEqual(tolerance);
+  });
+
+  it("holds across a range of input sizes", () => {
+    const reserveIn = 250_000_000_000n;
+    const reserveOut = 250_000_000_000n;
+    const pool = makePool(reserveIn.toString(), reserveOut.toString());
+    const k = reserveIn * reserveOut;
+
+    for (const inputAmount of [1n, 1_000n, 1_000_000n, 50_000_000_000n]) {
+      const result = estimateSwapOutput(pool, inputAmount.toString(), "XLM");
+      const newReserveIn = reserveIn + inputAmount;
+      const newReserveOut = reserveOut - BigInt(result.outputAmount);
+      const newK = newReserveIn * newReserveOut;
+
+      expect(newK).toBeLessThanOrEqual(k);
+      const drift = k - newK;
+      const tolerance = k / 1_000_000n;
+      expect(drift).toBeLessThanOrEqual(tolerance);
+    }
+  });
+
+  it("throws InsufficientLiquidityError for a swap larger than available reserves", () => {
+    const pool = makePool("1000", "1000");
+    expect(() => estimateSwapOutput(pool, "1000000", "XLM")).toThrow(
+      InsufficientLiquidityError,
+    );
+  });
+
+  it("returns 0 received and leaves reserves unchanged for a zero-amount swap", () => {
+    const reserveIn = 1_000_000n;
+    const reserveOut = 2_000_000n;
+    const pool = makePool(reserveIn.toString(), reserveOut.toString());
+
+    const result = estimateSwapOutput(pool, "0", "XLM");
+
+    expect(result.outputAmount).toBe("0");
+    // Reserves reported back on the pool object itself are untouched — the
+    // function is pure and never mutates its input.
+    expect(pool.reserves[0]!.amount).toBe(reserveIn.toString());
+    expect(pool.reserves[1]!.amount).toBe(reserveOut.toString());
+  });
+});

src/connectionPool.ts

@@ -34,6 +34,8 @@ interface PoolSlot {
   createdAt: number;
   lastUsedAt: number;
   lastSelectedAt: number;
+  /** Pending idle-recycle timer armed when this slot returns to inFlight === 0. */
+  idleTimer: ReturnType<typeof setTimeout> | null;
 }
 
 export interface PoolSlotStats {
@@ -232,10 +234,46 @@ export class ConnectionPool {
         slot.inFlight = Math.max(0, slot.inFlight - 1);
         if (error) slot.totalErrors += 1;
         slot.lastUsedAt = this.opts.now();
+
+        // Connection returned to the pool: (re)arm its idle-recycle timer so
+        // that a slot which stays idle for idleTimeoutMs gets closed even if
+        // nothing ever calls select() again to trigger lazy recycling
+        // (issue #360 follow-up).
+        if (slot.inFlight === 0) {
+          this._armIdleTimer(slotIdx, slot);
+        }
       },
     };
   }
 
+  /**
+   * Arm (replacing any existing) idle-recycle timer for the slot at `idx`.
+   * When the timer fires, if the slot is still idle and hasn't been used
+   * again in the meantime, it is recycled — the underlying connection is
+   * closed and replaced with a fresh one.
+   */
+  private _armIdleTimer(idx: number, slot: PoolSlot): void {
+    if (slot.idleTimer) {
+      clearTimeout(slot.idleTimer);
+      slot.idleTimer = null;
+    }
+    const armedAt = slot.lastUsedAt;
+    slot.idleTimer = setTimeout(() => {
+      if (this.disposed) return;
+      const current = this.slots[idx];
+      if (!current || current !== slot) return;
+      if (current.inFlight !== 0 || current.lastUsedAt !== armedAt) {
+        // Slot was reused since the timer was armed — nothing to do.
+        return;
+      }
+      this._recycle(idx, this.opts.now());
+    }, this.opts.idleTimeoutMs);
+    // Don't keep the process alive solely for pool housekeeping.
+    if (typeof (slot.idleTimer as unknown as { unref?: () => void }).unref === "function") {
+      (slot.idleTimer as unknown as { unref: () => void }).unref();
+    }
+  }
+
   /**
    * Attribute an error to the slot owning the given server. Useful for
    * callers using the synchronous `select()` path that don't pair a release.
@@ -289,6 +327,12 @@ export class ConnectionPool {
   dispose(): void {
     if (this.disposed) return;
     this.disposed = true;
+    for (const slot of this.slots) {
+      if (slot.idleTimer) {
+        clearTimeout(slot.idleTimer);
+        slot.idleTimer = null;
+      }
+    }
     this.slots = [];
   }
 
@@ -311,11 +355,16 @@ export class ConnectionPool {
       createdAt,
       lastUsedAt: createdAt,
       lastSelectedAt: createdAt,
+      idleTimer: null,
     };
   }
 
   private _recycle(idx: number, now: number): PoolSlot {
     const old = this.slots[idx]!;
+    if (old.idleTimer) {
+      clearTimeout(old.idleTimer);
+      old.idleTimer = null;
+    }
     const fresh = this._createSlot(old.recycledCount + 1, now);
     this.slots[idx] = fresh;
     return fresh;

src/requestBatcher.ts

@@ -37,6 +37,9 @@ export class BatchedRpcClient {
     windowMs = 10,
     maxBatchSize = 20,
   ) {
+    if (maxBatchSize === 0) {
+      throw new RangeError("maxBatchSize must not be 0");
+    }
     this._fetchers = fetchers;
     this._windowMs = windowMs;
     this._maxBatchSize = maxBatchSize;
@@ -135,14 +138,15 @@ export class BatchedRpcClient {
 /** @deprecated Use BatchedRpcClient instead */
 export interface BatcherConfig {
   windowMs: number;
-  maxBatchSize: number;
+  /** Maximum items per batch. Defaults to unlimited (Infinity) when omitted. */
+  maxBatchSize?: number;
 }
 
 /** @deprecated Use BatchedRpcClient instead */
 export class RequestBatcher {
   private readonly _inner: BatchedRpcClient;
 
-  constructor(config: BatcherConfig = { windowMs: 10, maxBatchSize: 20 }) {
+  constructor(config: BatcherConfig = { windowMs: 10, maxBatchSize: Infinity }) {
     const stub: BatchFetchers = {
       fetchInvoice: async (id) => ({
         id,
@@ -157,7 +161,11 @@ export class RequestBatcher {
       fetchPaymentHistory: async () => [],
       fetchInvoiceExt: async () => ({ parentInvoiceId: null, cloneDepth: 0 }),
     };
-    this._inner = new BatchedRpcClient(stub, config.windowMs, config.maxBatchSize);
+    this._inner = new BatchedRpcClient(
+      stub,
+      config.windowMs,
+      config.maxBatchSize ?? Infinity,
+    );
   }
 
   async getInvoice(invoiceId: string): Promise<Invoice> {

src/scheduler.ts

@@ -40,6 +40,104 @@ function save(payments: ScheduledPayment[]): void {
   } catch { /* no-op */ }
 }
 
+// ---------------------------------------------------------------------------
+// Generic job scheduler — recurring (interval) jobs and one-shot jobs
+// (issue #685).
+// ---------------------------------------------------------------------------
+
+/** Handle returned by `schedule`/`once`, allowing the caller to cancel the job. */
+export interface JobHandle {
+  id: string;
+  cancel: () => void;
+}
+
+interface RecurringJob {
+  id: string;
+  intervalMs: number;
+  fn: () => void | Promise<void>;
+  timer: ReturnType<typeof setInterval>;
+}
+
+interface OneShotJob {
+  id: string;
+  fn: () => void | Promise<void>;
+  timer: ReturnType<typeof setTimeout>;
+}
+
+/**
+ * Lightweight in-process job scheduler supporting both recurring
+ * (interval-based) jobs via `schedule()` and one-shot, run-once-after-a-delay
+ * jobs via `once()`. The two APIs coexist and share the same job list.
+ */
+export class JobScheduler {
+  private _recurring = new Map<string, RecurringJob>();
+  private _oneShot = new Map<string, OneShotJob>();
+
+  /** Schedule a recurring job that runs every `intervalMs` milliseconds. */
+  schedule(intervalMs: number, fn: () => void | Promise<void>): JobHandle {
+    const id = randomUUID();
+    const timer = setInterval(() => {
+      void fn();
+    }, intervalMs);
+    this._recurring.set(id, { id, intervalMs, fn, timer });
+
+    return {
+      id,
+      cancel: () => this._cancelRecurring(id),
+    };
+  }
+
+  /**
+   * Schedule a job to run exactly once, `delayMs` milliseconds from now.
+   * Once it runs (or is cancelled), the job is removed from the scheduler's
+   * internal job list.
+   */
+  once(delayMs: number, fn: () => void | Promise<void>): JobHandle {
+    const id = randomUUID();
+    const timer = setTimeout(() => {
+      this._oneShot.delete(id);
+      void fn();
+    }, delayMs);
+    this._oneShot.set(id, { id, fn, timer });
+
+    return {
+      id,
+      cancel: () => this._cancelOneShot(id),
+    };
+  }
+
+  /** Cancel a job (recurring or one-shot) by id. No-op if it no longer exists. */
+  cancel(id: string): void {
+    this._cancelRecurring(id);
+    this._cancelOneShot(id);
+  }
+
+  /** Number of jobs currently scheduled (recurring + pending one-shot). */
+  get jobCount(): number {
+    return this._recurring.size + this._oneShot.size;
+  }
+
+  /** Cancel every scheduled job. */
+  clear(): void {
+    for (const id of [...this._recurring.keys()]) this._cancelRecurring(id);
+    for (const id of [...this._oneShot.keys()]) this._cancelOneShot(id);
+  }
+
+  private _cancelRecurring(id: string): void {
+    const job = this._recurring.get(id);
+    if (!job) return;
+    clearInterval(job.timer);
+    this._recurring.delete(id);
+  }
+
+  private _cancelOneShot(id: string): void {
+    const job = this._oneShot.get(id);
+    if (!job) return;
+    clearTimeout(job.timer);
+    this._oneShot.delete(id);
+  }
+}
+
 export class ScheduledPaymentManager {
   private _payments: ScheduledPayment[] = load();
   private _timers = new Map<string, ReturnType<typeof setTimeout>>();

test/connectionPool.test.ts

@@ -1,4 +1,4 @@
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
   ConnectionPool,
   DEFAULT_IDLE_TIMEOUT_MS,
@@ -222,6 +222,58 @@ describe("ConnectionPool (issue #360)", () => {
     expect(() => pool.acquire()).toThrow();
   });
 
+  it("recycles a returned connection once its own idle timer fires (issue #684)", () => {
+    vi.useFakeTimers();
+    try {
+      const pool = new ConnectionPool({
+        rpcUrl: "https://soroban-testnet.stellar.org",
+        poolSize: 1,
+        idleTimeoutMs: 100,
+      });
+
+      const lease = pool.acquire();
+      const before = lease.server;
+      lease.release();
+
+      expect(pool.getStats().recycledCount).toBe(0);
+
+      // Advance past idleTimeoutMs without ever calling select()/acquire()
+      // again — the timer armed on release() must fire on its own.
+      vi.advanceTimersByTime(150);
+
+      expect(pool.getStats().recycledCount).toBe(1);
+      const after = pool.select();
+      expect(after).not.toBe(before);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("does not recycle a connection reused before its idle timer fires", () => {
+    vi.useFakeTimers();
+    try {
+      const pool = new ConnectionPool({
+        rpcUrl: "https://soroban-testnet.stellar.org",
+        poolSize: 1,
+        idleTimeoutMs: 100,
+      });
+
+      const lease = pool.acquire();
+      lease.release();
+
+      vi.advanceTimersByTime(50);
+      // Reuse the slot before the idle timer fires.
+      const lease2 = pool.acquire();
+      lease2.release();
+
+      vi.advanceTimersByTime(50);
+      // Only 50ms idle since the second release — should not have recycled.
+      expect(pool.getStats().recycledCount).toBe(0);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("accepts the legacy positional constructor signature", () => {
     const pool = new ConnectionPool(
       "https://soroban-testnet.stellar.org",

test/requestBatcher.test.ts

@@ -208,6 +208,11 @@ describe("BatchedRpcClient", () => {
     batcher.clear();
   });
 
+  it("throws RangeError when maxBatchSize is 0", () => {
+    const { fetchers } = makeFetchers();
+    expect(() => new BatchedRpcClient(fetchers, 10, 0)).toThrow(RangeError);
+  });
+
   it("overflow of exactly maxBatchSize triggers immediate flush (no timer needed)", async () => {
     const flushOrder: string[] = [];
     const fetchers: BatchFetchers = {