diff --git a/src/__tests__/ammCalculator.test.ts b/src/__tests__/ammCalculator.test.ts new file mode 100644 index 0000000..b650da5 --- /dev/null +++ b/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()); + }); +}); diff --git a/src/connectionPool.ts b/src/connectionPool.ts index 40dbb1d..f87ab19 100644 --- a/src/connectionPool.ts +++ b/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 | 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; diff --git a/src/requestBatcher.ts b/src/requestBatcher.ts index 7b373dc..d4a9160 100644 --- a/src/requestBatcher.ts +++ b/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 { diff --git a/src/scheduler.ts b/src/scheduler.ts index 81ee5a6..6578340 100644 --- a/src/scheduler.ts +++ b/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; + timer: ReturnType; +} + +interface OneShotJob { + id: string; + fn: () => void | Promise; + timer: ReturnType; +} + +/** + * 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(); + private _oneShot = new Map(); + + /** Schedule a recurring job that runs every `intervalMs` milliseconds. */ + schedule(intervalMs: number, fn: () => void | Promise): 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): 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>(); diff --git a/test/connectionPool.test.ts b/test/connectionPool.test.ts index ab943e4..28f9329 100644 --- a/test/connectionPool.test.ts +++ b/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", diff --git a/test/requestBatcher.test.ts b/test/requestBatcher.test.ts index 87d2a97..58d2450 100644 --- a/test/requestBatcher.test.ts +++ b/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 = { diff --git a/test/scheduler.test.ts b/test/scheduler.test.ts new file mode 100644 index 0000000..b77559d --- /dev/null +++ b/test/scheduler.test.ts @@ -0,0 +1,87 @@ +import { describe, it, expect, vi } from "vitest"; +import { JobScheduler } from "../src/scheduler.js"; + +describe("JobScheduler", () => { + it("once() runs fn exactly once after delayMs and removes the job from the list", () => { + vi.useFakeTimers(); + try { + const scheduler = new JobScheduler(); + const fn = vi.fn(); + + scheduler.once(100, fn); + expect(scheduler.jobCount).toBe(1); + + vi.advanceTimersByTime(100); + expect(fn).toHaveBeenCalledTimes(1); + expect(scheduler.jobCount).toBe(0); + + // Further time passing must not trigger it again. + vi.advanceTimersByTime(1000); + expect(fn).toHaveBeenCalledTimes(1); + } finally { + vi.useRealTimers(); + } + }); + + it("cancel() on the handle before the delay expires prevents fn from running", () => { + vi.useFakeTimers(); + try { + const scheduler = new JobScheduler(); + const fn = vi.fn(); + + const handle = scheduler.once(100, fn); + vi.advanceTimersByTime(50); + handle.cancel(); + vi.advanceTimersByTime(100); + + expect(fn).not.toHaveBeenCalled(); + expect(scheduler.jobCount).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("once() coexists with recurring schedule() jobs", () => { + vi.useFakeTimers(); + try { + const scheduler = new JobScheduler(); + const recurringFn = vi.fn(); + const oneShotFn = vi.fn(); + + scheduler.schedule(50, recurringFn); + scheduler.once(120, oneShotFn); + expect(scheduler.jobCount).toBe(2); + + vi.advanceTimersByTime(150); + + expect(recurringFn.mock.calls.length).toBeGreaterThanOrEqual(2); + expect(oneShotFn).toHaveBeenCalledTimes(1); + // The recurring job stays scheduled; only the one-shot job is removed. + expect(scheduler.jobCount).toBe(1); + + scheduler.clear(); + expect(scheduler.jobCount).toBe(0); + } finally { + vi.useRealTimers(); + } + }); + + it("schedule() cancel handle stops future recurring runs", () => { + vi.useFakeTimers(); + try { + const scheduler = new JobScheduler(); + const fn = vi.fn(); + + const handle = scheduler.schedule(50, fn); + vi.advanceTimersByTime(120); + expect(fn.mock.calls.length).toBeGreaterThanOrEqual(2); + + handle.cancel(); + const callsAtCancel = fn.mock.calls.length; + vi.advanceTimersByTime(200); + expect(fn.mock.calls.length).toBe(callsAtCancel); + } finally { + vi.useRealTimers(); + } + }); +});