fix: price impact scale normalization, batch concurrency/rate-limits, and anomaly NaN threshold

src/ammCalculator.ts

@@ -438,13 +438,16 @@ function computePriceImpact(spotPrice: string, effectivePrice: string): string {
 
   const spotScaled = spot.int * spot.scale + spot.frac;
   const effectiveScaled = effective.int * effective.scale + effective.frac;
+  const commonScale = spot.scale > effective.scale ? spot.scale : effective.scale;
+  const spotCommon = spotScaled * (commonScale / spot.scale);
+  const effectiveCommon = effectiveScaled * (commonScale / effective.scale);
 
-  if (spotScaled === 0n) return "0.00";
+  if (spotCommon === 0n) return "0.00";
 
   // (spot - effective) / spot * 100 with 4 decimal places of precision
   const SCALE = 10000n;
-  const numerator = (spotScaled - effectiveScaled) * SCALE * 100n;
-  const denominator = spotScaled;
+  const numerator = (spotCommon - effectiveCommon) * SCALE * 100n;
+  const denominator = spotCommon;
 
   if (numerator <= 0n) return "0.00";
 

src/anomalyDetector.ts

@@ -77,7 +77,11 @@ export class AnomalyDetector {
     this.rapidCycleSeconds = options.rapidCycleSeconds ?? 300;
     this.maxAmountVariance = options.maxAmountVariance ?? 0.8;
     this.sensitivityThreshold = options.sensitivityThreshold ?? 0.8;
-    if (this.sensitivityThreshold <= 0 || this.sensitivityThreshold > 1) {
+    if (
+      !Number.isFinite(this.sensitivityThreshold) ||
+      this.sensitivityThreshold <= 0 ||
+      this.sensitivityThreshold > 1
+    ) {
       throw new RangeError("sensitivityThreshold must be in the range (0, 1]");
     }
     this.now = options.now ?? (() => Math.floor(Date.now() / 1000));

src/invoiceBatchProcessor.ts

@@ -53,9 +53,14 @@ function isRateLimitError(error: unknown): boolean {
   return error instanceof Error && /429|rate.?limit|too many requests/i.test(error.message);
 }
 
+function isValidPauseMs(value: number): boolean {
+  return Number.isInteger(value) && value >= 0;
+}
+
 function retryAfterMs(error: unknown, fallbackMs: number): number {
   const withRetryAfter = error as { retryAfterMs?: number } | undefined;
-  return typeof withRetryAfter?.retryAfterMs === "number" ? withRetryAfter.retryAfterMs : fallbackMs;
+  const candidate = withRetryAfter?.retryAfterMs;
+  return typeof candidate === "number" && isValidPauseMs(candidate) ? candidate : fallbackMs;
 }
 
 export class InvoiceBatchProcessor {
@@ -76,14 +81,29 @@ export class InvoiceBatchProcessor {
     const maxConcurrent = config.maxConcurrent ?? DEFAULT_MAX_CONCURRENT;
     const rateLimitPauseMs = config.rateLimitPauseMs ?? DEFAULT_RATE_LIMIT_PAUSE_MS;
 
+    if (!Number.isInteger(maxConcurrent) || maxConcurrent < 1) {
+      throw new RangeError("maxConcurrent must be a positive integer");
+    }
+
+    if (!isValidPauseMs(rateLimitPauseMs)) {
+      throw new RangeError("rateLimitPauseMs must be a non-negative integer");
+    }
+
     let cursor = 0;
     let pausedUntil = 0;
     let slotSeq = 0;
     const inFlight = new Map<number, Promise<{ slot: number; result: BatchInvoiceResult }>>();
 
+    const waitForGlobalPause = async (): Promise<void> => {
+      while (true) {
+        const wait = pausedUntil - Date.now();
+        if (wait <= 0) return;
+        await sleep(wait);
+      }
+    };
+
     const runOne = async (invoiceId: string): Promise<BatchInvoiceResult> => {
-      const wait = pausedUntil - Date.now();
-      if (wait > 0) await sleep(wait);
+      await waitForGlobalPause();
 
       const amount = config.amounts[invoiceId];
       if (amount === undefined) {
@@ -103,7 +123,8 @@ export class InvoiceBatchProcessor {
         return result;
       } catch (error) {
         if (isRateLimitError(error)) {
-          pausedUntil = Date.now() + retryAfterMs(error, rateLimitPauseMs);
+          const nextPausedUntil = Date.now() + retryAfterMs(error, rateLimitPauseMs);
+          pausedUntil = Math.max(pausedUntil, nextPausedUntil);
         }
         const result: BatchInvoiceResult = {
           invoiceId,
@@ -125,7 +146,8 @@ export class InvoiceBatchProcessor {
       );
     };
 
-    for (let i = 0; i < maxConcurrent; i++) launch();
+    const initialLaunches = Math.min(maxConcurrent, invoiceIds.length);
+    for (let i = 0; i < initialLaunches; i++) launch();
 
     while (inFlight.size > 0) {
       const { slot, result } = await Promise.race(inFlight.values());

test/ammCalculator.test.ts

@@ -45,6 +45,20 @@ describe("estimateSwapOutput", () => {
     expect(parseFloat(result.spotPrice)).toBeCloseTo(1.0);
   });
 
+  it("normalizes integer and fractional prices before computing impact", () => {
+    const pool = makePool([
+      { asset: ASSET_X, amount: "1000" },
+      { asset: ASSET_Y, amount: "1000" },
+    ]);
+
+    const result = estimateSwapOutput(pool, "100", ASSET_X);
+
+    expect(result.outputAmount).toBe("91");
+    expect(result.spotPrice).toBe("1");
+    expect(result.effectivePrice).toBe("0.91");
+    expect(result.priceImpactPercent).toBe("9.00");
+  });
+
   it("returns zero output and zero price impact for zero input", () => {
     const result = estimateSwapOutput(POOL, "0", ASSET_X);
     expect(result.outputAmount).toBe("0");

test/anomalyDetector.nan.test.ts

@@ -0,0 +1,14 @@
+import { describe, expect, it } from "vitest";
+import { AnomalyDetector } from "../src/anomalyDetector.js";
+
+describe("AnomalyDetector sensitivityThreshold validation", () => {
+  it("rejects NaN instead of silently disabling score alerts", () => {
+    expect(() => new AnomalyDetector({ sensitivityThreshold: Number.NaN })).toThrow(
+      RangeError
+    );
+  });
+
+  it("still accepts finite values at the upper boundary", () => {
+    expect(() => new AnomalyDetector({ sensitivityThreshold: 1 })).not.toThrow();
+  });
+});

test/invoiceBatchProcessor.concurrency.test.ts

@@ -0,0 +1,49 @@
+import { describe, it, expect, vi } from "vitest";
+import { InvoiceBatchProcessor } from "../src/invoiceBatchProcessor.js";
+import type { InvoicePaymentSubmitter } from "../src/invoiceBatchProcessor.js";
+
+describe("InvoiceBatchProcessor – concurrency configuration", () => {
+  it.each([0, -1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY])(
+    "rejects invalid maxConcurrent=%s before submitting",
+    async (maxConcurrent) => {
+      const submitPayment = vi.fn().mockResolvedValue({ txHash: "tx-inv1" });
+      const processor = new InvoiceBatchProcessor(
+        { submitPayment } as InvoicePaymentSubmitter,
+      );
+
+      await expect(
+        processor.processAll(["inv1"], {
+          payer: "GPAYER",
+          amounts: { inv1: 1n },
+          maxConcurrent,
+        }),
+      ).rejects.toThrow(new RangeError("maxConcurrent must be a positive integer"));
+
+      expect(submitPayment).not.toHaveBeenCalled();
+    },
+  );
+
+  it("caps initial launch attempts at the number of invoices", async () => {
+    const submitPayment = vi
+      .fn()
+      .mockImplementation(async ({ invoiceId }: { invoiceId: string }) => ({
+        txHash: `tx-${invoiceId}`,
+      }));
+    const processor = new InvoiceBatchProcessor(
+      { submitPayment } as InvoicePaymentSubmitter,
+    );
+
+    const { succeeded, failed } = await processor.processAll(
+      ["inv1", "inv2"],
+      {
+        payer: "GPAYER",
+        amounts: { inv1: 1n, inv2: 1n },
+        maxConcurrent: Number.MAX_SAFE_INTEGER,
+      },
+    );
+
+    expect(succeeded).toHaveLength(2);
+    expect(failed).toHaveLength(0);
+    expect(submitPayment).toHaveBeenCalledTimes(2);
+  });
+});

test/invoiceBatchProcessor.rateLimit.test.ts

@@ -0,0 +1,142 @@
+import { describe, it, expect, vi } from "vitest";
+import { InvoiceBatchProcessor } from "../src/invoiceBatchProcessor.js";
+import type { InvoicePaymentSubmitter } from "../src/invoiceBatchProcessor.js";
+
+function rateLimitError(retryAfterMs: number): Error & { retryAfterMs: number } {
+  return Object.assign(new Error("429 rate limit"), { retryAfterMs });
+}
+
+describe("InvoiceBatchProcessor – rate-limit authority", () => {
+  it.each([-1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY])(
+    "rejects invalid rateLimitPauseMs=%s before submitting",
+    async (rateLimitPauseMs) => {
+      const submitPayment = vi.fn().mockResolvedValue({ txHash: "tx-inv1" });
+      const processor = new InvoiceBatchProcessor(
+        { submitPayment } as InvoicePaymentSubmitter,
+      );
+
+      await expect(
+        processor.processAll(["inv1"], {
+          payer: "GPAYER",
+          amounts: { inv1: 1n },
+          rateLimitPauseMs,
+        }),
+      ).rejects.toThrow(new RangeError("rateLimitPauseMs must be a non-negative integer"));
+
+      expect(submitPayment).not.toHaveBeenCalled();
+    },
+  );
+
+  it.each([-1, 1.5, Number.NaN, Number.POSITIVE_INFINITY, Number.NEGATIVE_INFINITY])(
+    "falls back when a 429 exposes malformed retryAfterMs=%s",
+    async (retryAfterMs) => {
+      vi.useFakeTimers();
+      try {
+        const submitPayment = vi
+          .fn()
+          .mockRejectedValueOnce(rateLimitError(retryAfterMs))
+          .mockResolvedValueOnce({ txHash: "tx-inv2" });
+        const processor = new InvoiceBatchProcessor(
+          { submitPayment } as InvoicePaymentSubmitter,
+        );
+
+        const run = processor.processAll(["inv1", "inv2"], {
+          payer: "GPAYER",
+          amounts: { inv1: 1n, inv2: 1n },
+          maxConcurrent: 1,
+          rateLimitPauseMs: 100,
+        });
+
+        await vi.advanceTimersByTimeAsync(0);
+        expect(submitPayment).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(99);
+        expect(submitPayment).toHaveBeenCalledTimes(1);
+        await vi.advanceTimersByTimeAsync(1);
+        await run;
+        expect(submitPayment).toHaveBeenCalledTimes(2);
+      } finally {
+        vi.useRealTimers();
+      }
+    },
+  );
+
+  it("does not let a shorter later 429 reduce an existing global pause", async () => {
+    vi.useFakeTimers();
+    try {
+      const submitPayment = vi.fn().mockImplementation(
+        ({ invoiceId }: { invoiceId: string }) => {
+          if (invoiceId === "inv1") {
+            return Promise.reject(rateLimitError(1_000));
+          }
+          if (invoiceId === "inv2") {
+            return new Promise((_resolve, reject) => {
+              setTimeout(() => reject(rateLimitError(100)), 50);
+            });
+          }
+          return Promise.resolve({ txHash: `tx-${invoiceId}` });
+        },
+      );
+      const processor = new InvoiceBatchProcessor(
+        { submitPayment } as InvoicePaymentSubmitter,
+      );
+
+      const run = processor.processAll(["inv1", "inv2", "inv3", "inv4"], {
+        payer: "GPAYER",
+        amounts: { inv1: 1n, inv2: 1n, inv3: 1n, inv4: 1n },
+        maxConcurrent: 2,
+      });
+
+      await vi.advanceTimersByTimeAsync(0);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(200);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(799);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(1);
+      await run;
+      expect(submitPayment).toHaveBeenCalledTimes(4);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  it("re-checks a shared pause when another in-flight 429 extends it", async () => {
+    vi.useFakeTimers();
+    try {
+      const submitPayment = vi.fn().mockImplementation(
+        ({ invoiceId }: { invoiceId: string }) => {
+          if (invoiceId === "inv1") {
+            return Promise.reject(rateLimitError(100));
+          }
+          if (invoiceId === "inv2") {
+            return new Promise((_resolve, reject) => {
+              setTimeout(() => reject(rateLimitError(1_000)), 50);
+            });
+          }
+          return Promise.resolve({ txHash: `tx-${invoiceId}` });
+        },
+      );
+      const processor = new InvoiceBatchProcessor(
+        { submitPayment } as InvoicePaymentSubmitter,
+      );
+
+      const run = processor.processAll(["inv1", "inv2", "inv3"], {
+        payer: "GPAYER",
+        amounts: { inv1: 1n, inv2: 1n, inv3: 1n },
+        maxConcurrent: 2,
+      });
+
+      await vi.advanceTimersByTimeAsync(0);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(100);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(949);
+      expect(submitPayment).toHaveBeenCalledTimes(2);
+      await vi.advanceTimersByTimeAsync(1);
+      await run;
+      expect(submitPayment).toHaveBeenCalledTimes(3);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+});