diff --git a/src/claimableBalanceFallback.ts b/src/claimableBalanceFallback.ts index ed4055f..b10f037 100644 --- a/src/claimableBalanceFallback.ts +++ b/src/claimableBalanceFallback.ts @@ -263,6 +263,12 @@ export interface ClaimableBalanceLifecycleEventMap { export interface ClaimableBalanceLifecycleConfig { /** Polling interval in milliseconds. Default: 10_000 (10s). */ pollIntervalMs?: number; + /** + * Time-to-live for tracked entries in milliseconds. Entries older than + * this are considered stale and will be removed by {@link ClaimableBalanceLifecycle.pruneExpired}. + * Default: 86_400_000 (24 hours). + */ + ttlMs?: number; } /** @@ -280,10 +286,20 @@ export interface ClaimableBalanceLifecycleConfig { * lifecycle.start(); * ``` */ +/** @internal Extended record stored inside the lifecycle manager. */ +interface TrackedEntry { + record: ClaimableBalanceRecord; + /** Unix epoch ms when this entry was registered via {@link ClaimableBalanceLifecycle.track}. */ + trackedAt: number; + /** TTL override for this specific entry (ms). Falls back to the manager default. */ + ttlMs: number; +} + export class ClaimableBalanceLifecycle extends TypedEventEmitter { private readonly server: Horizon.Server; private readonly pollIntervalMs: number; - private tracked: Map = new Map(); + private readonly _defaultTtlMs: number; + private tracked: Map = new Map(); private pollTimer: ReturnType | null = null; private _running = false; @@ -291,6 +307,7 @@ export class ClaimableBalanceLifecycle extends TypedEventEmitter= entry.ttlMs) { + this.tracked.delete(id); + removed += 1; + } + } + return removed; } /** @@ -344,7 +393,7 @@ export class ClaimableBalanceLifecycle extends TypedEventEmitter e.record); } /** @@ -361,13 +410,14 @@ export class ClaimableBalanceLifecycle extends TypedEventEmitter { try { - const record = this.tracked.get(balanceId); - if (!record) { + const entry = this.tracked.get(balanceId); + if (!entry) { throw new ClaimableBalanceLifecycleError( `Balance ${balanceId} is not tracked`, balanceId, ); } + const record = entry.record; const keypair = Keypair.fromSecret(claimantSecret); const account = await this.server.loadAccount(keypair.publicKey()); @@ -432,7 +482,8 @@ export class ClaimableBalanceLifecycle extends TypedEventEmitter { if (!this._running) return; - for (const [balanceId, record] of this.tracked.entries()) { + for (const [balanceId, entry] of this.tracked.entries()) { + const record = entry.record; try { const fresh = await this.server .claimableBalances() diff --git a/src/queue.ts b/src/queue.ts index e45bc3e..622f256 100644 --- a/src/queue.ts +++ b/src/queue.ts @@ -8,14 +8,31 @@ import { signTransaction } from "./wallet.js"; import type { TxResult } from "./client.js"; import { QueueFailedError } from "./errors.js"; -/** Transaction queue for serialized submission. */ +/** An item waiting in the priority queue. */ +interface QueueItem { + /** Higher priority items are dequeued first. Default is 0. */ + priority: number; + /** Insertion order index, used to preserve FIFO among equal-priority items. */ + seq: number; + operation: (account: Account) => Promise<{ txHash: string; returnValue: unknown }>; + resolve: (result: TxResult) => void; + reject: (error: unknown) => void; +} + +/** Transaction queue for serialized submission with optional priority ordering. */ export class TxQueue { private server: SorobanRpc.Server; private networkPassphrase: string; private sourceAddress: string; - private queue: Promise = Promise.resolve({ txHash: "" }); private failed = false; + /** Pending items sorted by priority (desc) then insertion order (asc). */ + private items: QueueItem[] = []; + /** Monotonically increasing sequence counter for FIFO tie-breaking. */ + private _seq = 0; + /** Whether the drain loop is currently running. */ + private _draining = false; + constructor( server: SorobanRpc.Server, networkPassphrase: string, @@ -29,35 +46,101 @@ export class TxQueue { /** * Enqueue an operation for sequential execution. * - * @param operation - The operation to execute - * @returns Promise resolving to transaction result + * @param operation - The operation to execute. + * @param priority - Higher values are processed first; equal-priority items + * are processed FIFO. Default: 0. + * @returns Promise resolving to transaction result. */ async enqueue( operation: ( account: Account - ) => Promise<{ txHash: string; returnValue: unknown }> + ) => Promise<{ txHash: string; returnValue: unknown }>, + priority = 0 ): Promise { if (this.failed) { throw new QueueFailedError(); } - this.queue = this.queue.then(async () => { - try { - const account = await this.server.getAccount(this.sourceAddress); - const result = await operation(account); - return { txHash: result.txHash }; - } catch (error) { - this.failed = true; - throw error; - } + return new Promise((resolve, reject) => { + const item: QueueItem = { + priority, + seq: this._seq++, + operation, + resolve, + reject, + }; + this._insert(item); + // Kick off the drain loop if it isn't already running. + void this._drain(); }); + } - return this.queue; + /** + * Return the next item that would be dequeued without removing it. + * Returns `undefined` when the queue is empty. + */ + peek(): { priority: number } | undefined { + const head = this.items[0]; + if (!head) return undefined; + return { priority: head.priority }; } - /** Clear the queue and reset state. */ + /** Clear the queue, reject all pending items, and reset state. */ clear(): void { - this.queue = Promise.resolve({ txHash: "" }); + const pending = this.items.splice(0); + for (const item of pending) { + item.reject(new QueueFailedError()); + } this.failed = false; + this._draining = false; + } + + // ------------------------------------------------------------------------- + // Internal helpers + // ------------------------------------------------------------------------- + + /** Insert an item in sorted order: higher priority first, FIFO on tie. */ + private _insert(item: QueueItem): void { + let lo = 0; + let hi = this.items.length; + while (lo < hi) { + const mid = (lo + hi) >>> 1; + const cand = this.items[mid]!; + // Sorted descending by priority, then ascending by seq + if ( + cand.priority > item.priority || + (cand.priority === item.priority && cand.seq < item.seq) + ) { + lo = mid + 1; + } else { + hi = mid; + } + } + this.items.splice(lo, 0, item); + } + + /** Sequential drain loop — processes items one at a time. */ + private async _drain(): Promise { + if (this._draining) return; + this._draining = true; + + while (this.items.length > 0 && !this.failed) { + const item = this.items.shift()!; + try { + const account = await this.server.getAccount(this.sourceAddress); + const result = await item.operation(account); + item.resolve({ txHash: result.txHash }); + } catch (error) { + this.failed = true; + item.reject(error); + // Reject all remaining items + const remaining = this.items.splice(0); + for (const r of remaining) { + r.reject(new QueueFailedError()); + } + } + } + + this._draining = false; } -} \ No newline at end of file +} diff --git a/src/retryEngine.ts b/src/retryEngine.ts index e39645b..840dfc1 100644 --- a/src/retryEngine.ts +++ b/src/retryEngine.ts @@ -9,6 +9,16 @@ export interface RetryStrategy { jitterMs?: number; } +export interface RetryEngineOptions { + /** + * Multiplicative jitter applied to each computed delay. + * Each delay is multiplied by a random value in [1 - jitterFactor, 1 + jitterFactor]. + * Must be in the range [0, 1]. Set to 0 to disable jitter entirely. + * @default 0.2 + */ + jitterFactor?: number; +} + export interface RetryConfig { transient: RetryStrategy; rateLimit: RetryStrategy; @@ -47,11 +57,19 @@ function sleep(ms: number): Promise { export class RetryEngine { private _consecutiveTransientFailures = 0; private _circuitOpenedAt: number | null = null; + private readonly _jitterFactor: number; constructor( private readonly config: RetryConfig, - private readonly telemetry: TelemetryCollector - ) {} + private readonly telemetry: TelemetryCollector, + options: RetryEngineOptions = {} + ) { + const jf = options.jitterFactor ?? 0.2; + if (jf < 0 || jf > 1) { + throw new Error(`jitterFactor must be in [0, 1], got ${jf}`); + } + this._jitterFactor = jf; + } get isCircuitOpen(): boolean { if (this._circuitOpenedAt === null) return false; @@ -110,10 +128,15 @@ export class RetryEngine { break; } - const delay = + const baseDelay = strategy.initialDelayMs * strategy.backoffMultiplier ** (attempt - 1) + (strategy.jitterMs ? Math.random() * strategy.jitterMs : 0); + const delay = + this._jitterFactor === 0 + ? baseDelay + : baseDelay * (1 - this._jitterFactor + Math.random() * 2 * this._jitterFactor); + await sleep(delay); } } diff --git a/src/types.ts b/src/types.ts index ee300a5..565cdf6 100644 --- a/src/types.ts +++ b/src/types.ts @@ -2080,3 +2080,30 @@ export interface SubentryCapacityError { /** The capacity result that triggered this error. */ capacityResult: SubentryCapacityResult; } + +// --------------------------------------------------------------------------- +// Claimable Balance Lifecycle Types +// --------------------------------------------------------------------------- + +/** Lifecycle status of a tracked claimable balance. */ +export type ClaimableBalanceStatus = "created" | "claimed" | "expired"; + +/** A claimable balance record tracked by {@link ClaimableBalanceLifecycle}. */ +export interface ClaimableBalanceRecord { + /** Stellar claimable balance ID (e.g. `00000000…`). */ + balanceId: string; + /** Stellar address of the account that can claim this balance. */ + claimant: string; + /** Asset descriptor: `"native"` for XLM, `"CODE:ISSUER"` for issued assets. */ + asset: string; + /** Human-readable amount string (e.g. `"12.5000000"`). */ + amount: string; + /** Current lifecycle status. */ + status: ClaimableBalanceStatus; + /** Unix epoch ms when the balance was created / first tracked. */ + createdAt: number; + /** Unix epoch ms when the balance was claimed, or `null` if not yet claimed. */ + claimedAt: number | null; + /** Ledger sequence after which the predicate expires (optional). */ + predicateExpiryLedger?: number; +} diff --git a/src/webhookValidator.ts b/src/webhookValidator.ts index 6f6b436..2861ca8 100644 --- a/src/webhookValidator.ts +++ b/src/webhookValidator.ts @@ -1,7 +1,28 @@ import { ValidationError } from "./errors.js"; +import { createHmac, timingSafeEqual } from "crypto"; const textEncoder = new TextEncoder(); +// --------------------------------------------------------------------------- +// Error +// --------------------------------------------------------------------------- + +/** + * Thrown by {@link validateWebhook} when the request's HMAC-SHA256 signature + * does not match the computed signature. + */ +export class WebhookSignatureError extends Error { + constructor(message = "Webhook signature verification failed") { + super(message); + this.name = "WebhookSignatureError"; + Object.setPrototypeOf(this, new.target.prototype); + } +} + +// --------------------------------------------------------------------------- +// Internal helpers +// --------------------------------------------------------------------------- + function normalizeHex(hex: string): string { return hex.toLowerCase(); } @@ -59,11 +80,14 @@ async function computeHmacSha256(secret: string, message: string): Promise`. + */ +const SIGNATURE_PREFIX = "hmac-sha256="; + +/** + * Verify the HMAC-SHA256 signature attached to a webhook delivery. + * + * The signature is expected in the `X-Split-Signature` header using the + * format `hmac-sha256=`. The HMAC is computed over the raw + * request body (bytes) so that JSON key ordering is preserved exactly as + * the sender signed it. + * + * Pass `secret = null` to skip verification entirely (opt-out mode). + * + * @param payload - Parsed request body (used only for type-checking; the + * HMAC is verified against `rawBody`). + * @param rawBody - The verbatim request body bytes / string received over + * the wire. Must match what the sender signed. + * @param secret - Shared HMAC secret. Pass `null` to skip verification. + * @param signature - Value of the `X-Split-Signature` header. + * + * @throws {WebhookSignatureError} When the computed HMAC does not match the + * provided signature. + */ +export function validateWebhook( + payload: unknown, + rawBody: string | Uint8Array, + secret: string | null, + signature: string +): void { + // Opt-out: skip verification when secret is explicitly null. + if (secret === null) { + return; + } + + // Strip the "hmac-sha256=" prefix if present. + const hexDigest = signature.startsWith(SIGNATURE_PREFIX) + ? signature.slice(SIGNATURE_PREFIX.length) + : signature; + + const body = + rawBody instanceof Uint8Array + ? rawBody + : Buffer.from(rawBody, "utf8"); + + const expected = createHmac("sha256", secret).update(body).digest(); + + let provided: Buffer; + try { + provided = Buffer.from(hexDigest, "hex"); + } catch { + throw new WebhookSignatureError("Webhook signature header is not valid hex"); + } + + if (provided.length === 0) { + throw new WebhookSignatureError("Webhook signature header is empty or not valid hex"); + } + + // Use Node.js timingSafeEqual to prevent timing attacks. + const match = + expected.length === provided.length && + timingSafeEqual(expected, provided); + + if (!match) { + throw new WebhookSignatureError(); + } +} diff --git a/test/claimableBalanceFallback.ttl.test.ts b/test/claimableBalanceFallback.ttl.test.ts new file mode 100644 index 0000000..4f81177 --- /dev/null +++ b/test/claimableBalanceFallback.ttl.test.ts @@ -0,0 +1,206 @@ +/** + * Unit tests for the TTL / pruneExpired() feature added to + * ClaimableBalanceLifecycle (src/claimableBalanceFallback.ts). + */ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { ClaimableBalanceLifecycle } from "../src/claimableBalanceFallback.js"; +import type { ClaimableBalanceRecord } from "../src/types.js"; + +// --------------------------------------------------------------------------- +// Minimal Horizon.Server stub (we don't test polling here) +// --------------------------------------------------------------------------- +function makeMockServer() { + return { + claimableBalances: vi.fn().mockReturnValue({ + claimant: vi.fn().mockReturnValue({ + call: vi.fn().mockResolvedValue({ records: [] }), + }), + }), + loadAccount: vi.fn(), + submitTransaction: vi.fn(), + operations: vi.fn(), + } as unknown as import("@stellar/stellar-sdk").Horizon.Server; +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeRecord(id: string): ClaimableBalanceRecord { + return { + balanceId: id, + claimant: "GCLAIMANT000000000000000000000000000000000000000000000000", + asset: "native", + amount: "1.0000000", + status: "created", + createdAt: Date.now(), + claimedAt: null, + }; +} + +// --------------------------------------------------------------------------- +// TTL configuration +// --------------------------------------------------------------------------- + +describe("ClaimableBalanceLifecycle – TTL configuration", () => { + it("defaults to 24 h (86_400_000 ms)", () => { + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer()); + expect(lifecycle.defaultTtlMs).toBe(86_400_000); + }); + + it("accepts a custom ttlMs via constructor options", () => { + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 5_000 }); + expect(lifecycle.defaultTtlMs).toBe(5_000); + }); +}); + +// --------------------------------------------------------------------------- +// Creation timestamp and TTL stored per entry +// --------------------------------------------------------------------------- + +describe("ClaimableBalanceLifecycle – track() stores timestamp", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("stores the current time as trackedAt when an entry is inserted", () => { + vi.setSystemTime(1_000_000); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 60_000 }); + lifecycle.track(makeRecord("bal-1")); + + // The entry is visible in listTracked() + const tracked = lifecycle.listTracked(); + expect(tracked).toHaveLength(1); + expect(tracked[0]!.balanceId).toBe("bal-1"); + expect(lifecycle.trackedCount).toBe(1); + }); +}); + +// --------------------------------------------------------------------------- +// pruneExpired() +// --------------------------------------------------------------------------- + +describe("ClaimableBalanceLifecycle – pruneExpired()", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("returns 0 when no entries have expired", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 10_000 }); + lifecycle.track(makeRecord("a")); + lifecycle.track(makeRecord("b")); + + vi.advanceTimersByTime(5_000); // halfway through TTL + expect(lifecycle.pruneExpired()).toBe(0); + expect(lifecycle.trackedCount).toBe(2); + }); + + it("removes entries that have reached or exceeded their TTL", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 10_000 }); + lifecycle.track(makeRecord("a")); + lifecycle.track(makeRecord("b")); + + vi.advanceTimersByTime(10_000); // exactly at TTL boundary + const removed = lifecycle.pruneExpired(); + expect(removed).toBe(2); + expect(lifecycle.trackedCount).toBe(0); + }); + + it("only removes entries past TTL, leaving unexpired entries intact", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 30_000 }); + lifecycle.track(makeRecord("old")); // inserted at t=0 + + vi.advanceTimersByTime(20_000); // advance to t=20 s + + lifecycle.track(makeRecord("new")); // inserted at t=20 s, TTL expires at t=50 s + + vi.advanceTimersByTime(15_000); // advance to t=35 s → "old" is at 35 s (> 30 s), "new" is at 15 s + + const removed = lifecycle.pruneExpired(); + expect(removed).toBe(1); + expect(lifecycle.trackedCount).toBe(1); + expect(lifecycle.listTracked()[0]!.balanceId).toBe("new"); + }); + + it("returns 0 when the map is empty", () => { + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer()); + expect(lifecycle.pruneExpired()).toBe(0); + }); +}); + +// --------------------------------------------------------------------------- +// Auto-prune on insert +// --------------------------------------------------------------------------- + +describe("ClaimableBalanceLifecycle – auto-prune before insert", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("prunes stale entries automatically when track() is called", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 5_000 }); + + // Insert two entries + lifecycle.track(makeRecord("stale-1")); + lifecycle.track(makeRecord("stale-2")); + expect(lifecycle.trackedCount).toBe(2); + + // Advance past TTL + vi.advanceTimersByTime(6_000); + + // Inserting a new entry should prune the stale ones first + lifecycle.track(makeRecord("fresh")); + expect(lifecycle.trackedCount).toBe(1); + expect(lifecycle.listTracked()[0]!.balanceId).toBe("fresh"); + }); + + it("does not prune entries that are still within their TTL", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 60_000 }); + lifecycle.track(makeRecord("alive-1")); + lifecycle.track(makeRecord("alive-2")); + + vi.advanceTimersByTime(30_000); + + lifecycle.track(makeRecord("alive-3")); + expect(lifecycle.trackedCount).toBe(3); + }); +}); + +// --------------------------------------------------------------------------- +// Per-entry TTL override +// --------------------------------------------------------------------------- + +describe("ClaimableBalanceLifecycle – per-entry TTL override", () => { + beforeEach(() => vi.useFakeTimers()); + afterEach(() => vi.useRealTimers()); + + it("respects a per-entry TTL that is shorter than the manager default", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 60_000 }); + + lifecycle.track(makeRecord("short"), 5_000); // expires at t=5 s + lifecycle.track(makeRecord("long")); // expires at t=60 s + + vi.advanceTimersByTime(10_000); + + const removed = lifecycle.pruneExpired(); + expect(removed).toBe(1); + expect(lifecycle.listTracked()[0]!.balanceId).toBe("long"); + }); + + it("respects a per-entry TTL that is longer than the manager default", () => { + vi.setSystemTime(0); + const lifecycle = new ClaimableBalanceLifecycle(makeMockServer(), { ttlMs: 5_000 }); + + lifecycle.track(makeRecord("extended"), 60_000); // custom: 60 s + lifecycle.track(makeRecord("default")); // default: 5 s + + vi.advanceTimersByTime(10_000); + + const removed = lifecycle.pruneExpired(); + expect(removed).toBe(1); + expect(lifecycle.listTracked()[0]!.balanceId).toBe("extended"); + }); +}); diff --git a/test/queue.test.ts b/test/queue.test.ts new file mode 100644 index 0000000..73a973c --- /dev/null +++ b/test/queue.test.ts @@ -0,0 +1,283 @@ +/** + * Unit tests for TxQueue – priority ordering, FIFO tie-breaking, peek(), + * and the existing zero-priority FIFO contract. + */ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { TxQueue } from "../src/queue.js"; +import { QueueFailedError } from "../src/errors.js"; + +// --------------------------------------------------------------------------- +// Mock @stellar/stellar-sdk +// --------------------------------------------------------------------------- + +vi.mock("@stellar/stellar-sdk", async () => { + const actual = await vi.importActual("@stellar/stellar-sdk"); + return { + ...(actual as Record), + rpc: { + Server: vi.fn(), + }, + }; +}); + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +type MockServer = { + getAccount: ReturnType; +}; + +function makeMockServer(): MockServer { + return { + getAccount: vi.fn().mockResolvedValue({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + }), + }; +} + +function makeQueue(server: MockServer): TxQueue { + const { rpc } = require("@stellar/stellar-sdk"); + // Point the constructor to our mock — cast through unknown for DI + return new TxQueue( + server as unknown as import("@stellar/stellar-sdk").rpc.Server, + "Test SDF Network ; September 2015", + "GSOURCE000000000000000000000000000000000000000000000000000" + ); +} + +/** Returns a simple operation that records itself in `order` and resolves. */ +function makeOp(id: string, order: string[]) { + return async (_account: unknown) => { + order.push(id); + return { txHash: `tx-${id}`, returnValue: null }; + }; +} + +// --------------------------------------------------------------------------- +// Zero-priority FIFO contract (existing behaviour preserved) +// --------------------------------------------------------------------------- + +describe("TxQueue – zero-priority FIFO", () => { + it("processes operations in insertion order when all priorities are 0", async () => { + const server = makeMockServer(); + const queue = makeQueue(server); + const order: string[] = []; + + const p1 = queue.enqueue(makeOp("first", order)); + const p2 = queue.enqueue(makeOp("second", order)); + const p3 = queue.enqueue(makeOp("third", order)); + + await Promise.all([p1, p2, p3]); + expect(order).toEqual(["first", "second", "third"]); + }); + + it("resolves with the correct txHash", async () => { + const server = makeMockServer(); + const queue = makeQueue(server); + const result = await queue.enqueue(makeOp("a", [])); + expect(result.txHash).toBe("tx-a"); + }); +}); + +// --------------------------------------------------------------------------- +// Priority ordering +// --------------------------------------------------------------------------- + +describe("TxQueue – priority ordering", () => { + it("processes the highest-priority item first", async () => { + const server = makeMockServer(); + // Make getAccount block until we release it so all items can be enqueued + // before processing starts. + let resolveFirst!: () => void; + let firstCall = true; + server.getAccount = vi.fn().mockImplementation(() => { + if (firstCall) { + firstCall = false; + // The very first drain iteration will block here until we release it + return new Promise((r) => { resolveFirst = r; }).then(() => ({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + })); + } + return Promise.resolve({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + }); + }); + + const queue = makeQueue(server); + const order: string[] = []; + + // Enqueue low-priority first, then high before the first item finishes + const pLow = queue.enqueue(makeOp("low", order), 1); + const pHigh = queue.enqueue(makeOp("high", order), 10); + const pUrgent = queue.enqueue(makeOp("urgent", order), 100); + + // Release the first blocked getAccount so drain can proceed + resolveFirst(); + + await Promise.all([pLow, pHigh, pUrgent]); + + // After the first item ("low" was first in queue but then higher-priority items were added) + // the drain processes them in priority order: urgent → high → low + // However "low" was already dequeued and processing when high/urgent arrived. + // So the real sequence is: low (already running), then urgent, then high. + // This is the correct priority-queue behaviour: once dequeued, an item runs. + // Items still waiting are processed in priority order. + expect(order[0]).toBe("low"); // already dequeued + expect(order[1]).toBe("urgent"); + expect(order[2]).toBe("high"); + }); + + it("FIFO ordering preserved among equal-priority items", async () => { + const server = makeMockServer(); + const queue = makeQueue(server); + const order: string[] = []; + + // All at the same non-zero priority + const p1 = queue.enqueue(makeOp("alpha", order), 5); + const p2 = queue.enqueue(makeOp("beta", order), 5); + const p3 = queue.enqueue(makeOp("gamma", order), 5); + + await Promise.all([p1, p2, p3]); + expect(order).toEqual(["alpha", "beta", "gamma"]); + }); + + it("mixes zero and non-zero priorities correctly", async () => { + const server = makeMockServer(); + + let resolveFirst!: () => void; + let firstCall = true; + server.getAccount = vi.fn().mockImplementation(() => { + if (firstCall) { + firstCall = false; + return new Promise((r) => { resolveFirst = r; }).then(() => ({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + })); + } + return Promise.resolve({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + }); + }); + + const queue = makeQueue(server); + const order: string[] = []; + + const pNormal = queue.enqueue(makeOp("normal", order), 0); + const pPriority = queue.enqueue(makeOp("priority", order), 50); + + resolveFirst(); + await Promise.all([pNormal, pPriority]); + + // "normal" was already dequeued (first item), then "priority" runs next + expect(order[0]).toBe("normal"); + expect(order[1]).toBe("priority"); + }); +}); + +// --------------------------------------------------------------------------- +// peek() +// --------------------------------------------------------------------------- + +describe("TxQueue – peek()", () => { + it("returns undefined when the queue is empty", () => { + const server = makeMockServer(); + const queue = makeQueue(server); + expect(queue.peek()).toBeUndefined(); + }); + + it("reflects the priority of items waiting in the queue", async () => { + const server = makeMockServer(); + + // Capture the resolve function so we can unblock getAccount on demand + let unblockFirst!: (v: unknown) => void; + let callCount = 0; + server.getAccount = vi.fn().mockImplementation(() => { + callCount++; + if (callCount === 1) { + // First call blocks — this holds p1 in flight inside the drain loop + return new Promise((resolve) => { unblockFirst = resolve; }); + } + return Promise.resolve({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + }); + }); + + const queue2 = makeQueue(server); + const order: string[] = []; + + // p1 enqueued — drain loop starts and blocks on the first getAccount + const p1 = queue2.enqueue(makeOp("p1", order), 1); + + // Let the drain loop tick so it dequeues p1 and starts awaiting getAccount + await Promise.resolve(); + + // p2 arrives while p1 is in flight — it sits in items[] + const p2 = queue2.enqueue(makeOp("p2", order), 10); + + // peek() reflects the waiting items (p2), not the in-flight p1 + expect(queue2.peek()).toEqual({ priority: 10 }); + + // Unblock p1 and let both complete normally + unblockFirst({ + accountId: () => "GSOURCE", + sequenceNumber: () => "1", + incrementSequenceNumber: vi.fn(), + }); + + await Promise.all([p1, p2]); + expect(order).toEqual(["p1", "p2"]); + }); + + it("returns undefined after all items have been processed", async () => { + const server = makeMockServer(); + const queue = makeQueue(server); + await queue.enqueue(makeOp("x", [])); + // After processing, queue is empty + expect(queue.peek()).toBeUndefined(); + }); +}); + +// --------------------------------------------------------------------------- +// Failure propagation +// --------------------------------------------------------------------------- + +describe("TxQueue – failure propagation", () => { + it("throws QueueFailedError for new enqueues after a failure", async () => { + const server = makeMockServer(); + server.getAccount = vi.fn().mockRejectedValue(new Error("rpc down")); + + const queue = makeQueue(server); + await expect(queue.enqueue(makeOp("fail", []))).rejects.toThrow("rpc down"); + + await expect(queue.enqueue(makeOp("after-fail", []))).rejects.toThrow(QueueFailedError); + }); + + it("clear() resets the failed state", async () => { + const server = makeMockServer(); + server.getAccount = vi.fn() + .mockRejectedValueOnce(new Error("rpc down")) + .mockResolvedValue({ + accountId: () => "GSOURCE", sequenceNumber: () => "1", incrementSequenceNumber: vi.fn(), + }); + + const queue = makeQueue(server); + await expect(queue.enqueue(makeOp("fail", []))).rejects.toThrow("rpc down"); + + queue.clear(); + + const result = await queue.enqueue(makeOp("recovery", [])); + expect(result.txHash).toBe("tx-recovery"); + }); +}); diff --git a/test/retryEngine.jitter.test.ts b/test/retryEngine.jitter.test.ts new file mode 100644 index 0000000..75307f6 --- /dev/null +++ b/test/retryEngine.jitter.test.ts @@ -0,0 +1,211 @@ +/** + * Unit tests for the jitterFactor feature added to RetryEngine + * (src/retryEngine.ts). + */ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { RetryEngine } from "../src/retryEngine.js"; +import type { RetryConfig } from "../src/retryEngine.js"; +import { TelemetryCollector } from "../src/telemetryCollector.js"; + +const baseConfig: RetryConfig = { + transient: { maxAttempts: 3, initialDelayMs: 100, backoffMultiplier: 2 }, + rateLimit: { maxAttempts: 2, initialDelayMs: 50, backoffMultiplier: 1 }, + contract: { maxAttempts: 1, initialDelayMs: 0, backoffMultiplier: 1 }, + circuitBreakerThreshold: 10, + circuitResetMs: 500, +}; + +function makeEngine( + cfg: Partial = {}, + jitterFactor?: number +): { engine: RetryEngine; telemetry: TelemetryCollector } { + const telemetry = new TelemetryCollector(); + const engine = new RetryEngine( + { ...baseConfig, ...cfg }, + telemetry, + jitterFactor !== undefined ? { jitterFactor } : {} + ); + return { engine, telemetry }; +} + +beforeEach(() => vi.useFakeTimers()); +afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); +}); + +// --------------------------------------------------------------------------- +// Default jitterFactor = 0.2 +// --------------------------------------------------------------------------- + +describe("RetryEngine – jitterFactor default (0.2)", () => { + it("engine constructs without error when jitterFactor is not specified", () => { + expect(() => makeEngine()).not.toThrow(); + }); + + it("retries still succeed with default jitter applied", async () => { + const { engine } = makeEngine( + { transient: { maxAttempts: 3, initialDelayMs: 10, backoffMultiplier: 1 } }, + 0.2 + ); + const fn = vi.fn() + .mockRejectedValueOnce(new Error("transient")) + .mockResolvedValue("ok"); + + const promise = engine.execute(fn, "m"); + await vi.runAllTimersAsync(); + await expect(promise).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); +}); + +// --------------------------------------------------------------------------- +// jitterFactor = 0 (disabled) — verified via Math.random spy +// --------------------------------------------------------------------------- + +describe("RetryEngine – jitterFactor = 0 (no jitter)", () => { + it("delay equals base delay exactly when jitterFactor is 0", async () => { + // With jitterFactor=0 the formula is: delay = baseDelay (no multiplication). + // We verify by pinning Math.random to a non-neutral value and checking + // that the actual sleep duration is still exactly baseDelay. + // Since we can't easily intercept setTimeout (fake timers own it), we + // instead verify via Math.random never being called for the jitter path. + const randomSpy = vi.spyOn(Math, "random"); + + const { engine } = makeEngine( + { transient: { maxAttempts: 2, initialDelayMs: 50, backoffMultiplier: 1 } }, + 0 + ); + + const fn = vi.fn() + .mockRejectedValueOnce(new Error("transient")) + .mockResolvedValue("ok"); + + const promise = engine.execute(fn, "m"); + await vi.runAllTimersAsync(); + await promise; + + // Math.random must NOT be called for the jitter multiplication when jitterFactor=0 + // (it may still be called 0 times — that's what we're asserting) + expect(randomSpy).not.toHaveBeenCalled(); + }); +}); + +// --------------------------------------------------------------------------- +// jitterFactor range validation +// --------------------------------------------------------------------------- + +describe("RetryEngine – jitterFactor validation", () => { + it("throws when jitterFactor is below 0", () => { + const telemetry = new TelemetryCollector(); + expect( + () => new RetryEngine(baseConfig, telemetry, { jitterFactor: -0.1 }) + ).toThrow(/jitterFactor.*\[0.*1\]/i); + }); + + it("throws when jitterFactor exceeds 1", () => { + const telemetry = new TelemetryCollector(); + expect( + () => new RetryEngine(baseConfig, telemetry, { jitterFactor: 1.5 }) + ).toThrow(/jitterFactor.*\[0.*1\]/i); + }); + + it("accepts jitterFactor = 0 (boundary)", () => { + const telemetry = new TelemetryCollector(); + expect(() => new RetryEngine(baseConfig, telemetry, { jitterFactor: 0 })).not.toThrow(); + }); + + it("accepts jitterFactor = 1 (boundary)", () => { + const telemetry = new TelemetryCollector(); + expect(() => new RetryEngine(baseConfig, telemetry, { jitterFactor: 1 })).not.toThrow(); + }); +}); + +// --------------------------------------------------------------------------- +// Delay range: verified by controlling Math.random +// --------------------------------------------------------------------------- + +describe("RetryEngine – delay range with jitterFactor", () => { + it("delay equals base*(1-jf) when Math.random returns 0", async () => { + // With jitterFactor=0.3 and Math.random()=0: + // delay = base * (1 - 0.3 + 0 * 2 * 0.3) = base * 0.7 + vi.spyOn(Math, "random").mockReturnValue(0); + + const jitterFactor = 0.3; + const initialDelayMs = 100; + const { engine } = makeEngine( + { + transient: { maxAttempts: 2, initialDelayMs, backoffMultiplier: 1 }, + circuitBreakerThreshold: 10, + }, + jitterFactor + ); + + const fn = vi.fn() + .mockRejectedValueOnce(new Error("transient")) + .mockResolvedValue("ok"); + + // Capture the actual timer delay vitest schedules + const scheduledDelays: number[] = []; + // vi.advanceTimersByTimeAsync tracks timers — we verify the engine still + // completes, meaning the sleep fired at ≤ base*(1+jf). + // Since Math.random()=0, expected delay = 100 * (1 - 0.3) = 70ms + const promise = engine.execute(fn, "m"); + // Advance just 70ms — if jitter is applied correctly the sleep resolves + await vi.advanceTimersByTimeAsync(70); + await expect(promise).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it("delay equals base*(1+jf) when Math.random returns 1", async () => { + // With jitterFactor=0.3 and Math.random()=1: + // delay = base * (1 - 0.3 + 1 * 2 * 0.3) = base * 1.3 = 130ms + // Verify the sleep fires at ~130ms by advancing exactly that far. + vi.spyOn(Math, "random").mockReturnValue(1); + + const jitterFactor = 0.3; + const initialDelayMs = 100; + const { engine } = makeEngine( + { + transient: { maxAttempts: 2, initialDelayMs, backoffMultiplier: 1 }, + circuitBreakerThreshold: 10, + }, + jitterFactor + ); + + const fn = vi.fn() + .mockRejectedValueOnce(new Error("transient")) + .mockResolvedValue("ok"); + + const promise = engine.execute(fn, "m"); + // Advance to exactly 130ms — the sleep should have fired + await vi.advanceTimersByTimeAsync(130); + await expect(promise).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it("Math.random is called once per retry when jitterFactor > 0", async () => { + const randomSpy = vi.spyOn(Math, "random").mockReturnValue(0.5); + + const { engine } = makeEngine( + { + transient: { maxAttempts: 4, initialDelayMs: 10, backoffMultiplier: 1 }, + circuitBreakerThreshold: 10, + }, + 0.2 + ); + + const fn = vi.fn() + .mockRejectedValueOnce(new Error("t")) + .mockRejectedValueOnce(new Error("t")) + .mockRejectedValueOnce(new Error("t")) + .mockResolvedValue("ok"); + + const promise = engine.execute(fn, "m"); + await vi.runAllTimersAsync(); + await promise; + + // Three retries → Math.random called exactly 3 times (once per sleep) + expect(randomSpy).toHaveBeenCalledTimes(3); + }); +}); diff --git a/test/webhookValidator.hmac.test.ts b/test/webhookValidator.hmac.test.ts new file mode 100644 index 0000000..0ffbf21 --- /dev/null +++ b/test/webhookValidator.hmac.test.ts @@ -0,0 +1,129 @@ +/** + * Unit tests for the validateWebhook() HMAC enforcement added to + * src/webhookValidator.ts. + */ +import { describe, it, expect } from "vitest"; +import { createHmac } from "crypto"; +import { validateWebhook, WebhookSignatureError } from "../src/webhookValidator.js"; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function sign(body: string, secret: string): string { + return "hmac-sha256=" + createHmac("sha256", secret).update(body).digest("hex"); +} + +function signRaw(body: string, secret: string): string { + return createHmac("sha256", secret).update(body).digest("hex"); +} + +const SECRET = "super-secret-key"; +const BODY = JSON.stringify({ event: "payment", amount: 42 }); +const PAYLOAD = { event: "payment", amount: 42 }; + +// --------------------------------------------------------------------------- +// Valid signatures +// --------------------------------------------------------------------------- + +describe("validateWebhook – valid signatures", () => { + it("does not throw for a correct hmac-sha256= prefixed signature", () => { + const sig = sign(BODY, SECRET); + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, sig)).not.toThrow(); + }); + + it("does not throw for a bare hex digest (no prefix)", () => { + const sig = signRaw(BODY, SECRET); + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, sig)).not.toThrow(); + }); + + it("verifies correctly when rawBody is a Uint8Array", () => { + const bodyBytes = Buffer.from(BODY, "utf8"); + const sig = sign(BODY, SECRET); + expect(() => validateWebhook(PAYLOAD, bodyBytes, SECRET, sig)).not.toThrow(); + }); + + it("verifies different payload bodies independently", () => { + const body1 = JSON.stringify({ event: "refund" }); + const body2 = JSON.stringify({ event: "release" }); + const sig1 = sign(body1, SECRET); + const sig2 = sign(body2, SECRET); + + expect(() => validateWebhook({}, body1, SECRET, sig1)).not.toThrow(); + expect(() => validateWebhook({}, body2, SECRET, sig2)).not.toThrow(); + }); +}); + +// --------------------------------------------------------------------------- +// WebhookSignatureError — mismatched signatures +// --------------------------------------------------------------------------- + +describe("validateWebhook – signature mismatch throws WebhookSignatureError", () => { + it("throws WebhookSignatureError when signature is wrong", () => { + const wrongSig = sign(BODY, "wrong-secret"); + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, wrongSig)).toThrow(WebhookSignatureError); + }); + + it("throws when the body has been tampered", () => { + const sig = sign(BODY, SECRET); + const tamperedBody = JSON.stringify({ event: "payment", amount: 99 }); + expect(() => validateWebhook(PAYLOAD, tamperedBody, SECRET, sig)).toThrow(WebhookSignatureError); + }); + + it("throws when an empty signature is provided", () => { + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, "")).toThrow(WebhookSignatureError); + }); + + it("throws when the signature prefix is correct but digest is garbage", () => { + expect(() => + validateWebhook(PAYLOAD, BODY, SECRET, "hmac-sha256=notahexstring") + ).toThrow(WebhookSignatureError); + }); + + it("WebhookSignatureError is an instance of Error", () => { + try { + validateWebhook(PAYLOAD, BODY, SECRET, "bad"); + } catch (err) { + expect(err).toBeInstanceOf(Error); + expect(err).toBeInstanceOf(WebhookSignatureError); + expect((err as Error).name).toBe("WebhookSignatureError"); + } + }); +}); + +// --------------------------------------------------------------------------- +// Opt-out mode (secret = null) +// --------------------------------------------------------------------------- + +describe("validateWebhook – opt-out mode (secret = null)", () => { + it("does not throw when secret is null, regardless of signature", () => { + expect(() => validateWebhook(PAYLOAD, BODY, null, "garbage-sig")).not.toThrow(); + }); + + it("does not throw when secret is null and signature is empty", () => { + expect(() => validateWebhook(PAYLOAD, BODY, null, "")).not.toThrow(); + }); + + it("does not throw when secret is null and body is empty", () => { + expect(() => validateWebhook({}, "", null, "whatever")).not.toThrow(); + }); +}); + +// --------------------------------------------------------------------------- +// Timing-safe: same result regardless of secret length (smoke test) +// --------------------------------------------------------------------------- + +describe("validateWebhook – timing safety smoke test", () => { + it("rejects a signature computed with a different secret (no secret leakage)", () => { + const sig = sign(BODY, "other-secret"); + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, sig)).toThrow(WebhookSignatureError); + }); + + it("accepts the correct signature after rejecting an incorrect one", () => { + const badSig = sign(BODY, "wrong"); + const goodSig = sign(BODY, SECRET); + + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, badSig)).toThrow(WebhookSignatureError); + expect(() => validateWebhook(PAYLOAD, BODY, SECRET, goodSig)).not.toThrow(); + }); +});