From 001e5433058b340c9835756521e5b434879a4995 Mon Sep 17 00:00:00 2001 From: Josh Date: Thu, 23 Jul 2026 17:40:21 +0200 Subject: [PATCH 1/5] Fix Firo Open CryptoPay minimum fee rejecting Spark payments (#4305) * Fix Firo Open CryptoPay minimum fee rejecting Spark payments The Firo customer-facing Open CryptoPay minimum was taken from DFX's own payout fee rate (estimateSmartFee times the CPFP/default margin, ~2.068 sat/vB). Firo Spark transactions carry a protocol-fixed fee at the network relay minimum (~1 sat/vB) that the user cannot raise, so valid Spark payments were rejected. Use Firo's relay floor as the customer minimum (configurable via FIRO_MIN_FEE_RATE, default 1 sat/vB); the payout margin stays on DFX's own payout path. Bitcoin keeps its margin-based minimum because its fees are user-adjustable. * Decouple Bitcoin Open CryptoPay minimum from payout fee margin For consistency with the Firo fix, the customer-facing Open CryptoPay minimum for Bitcoin no longer derives from DFX's payout send rate (which carries a CPFP/default margin meant only for DFX's own outbound spends). Use the network's recommended next-block rate, floored at the relay minimum so the advertised minimum stays relayable. The payout margin remains on DFX's own payout/payin/dex paths, unchanged. * Derive Firo Open CryptoPay minimum from node rate instead of hardcoded floor The current OCP Firo deposit address is transparent, so a Stack Wallet payment is a Spark-spend to it whose fee sits at the relay floor and cannot be raised. Mirror the Bitcoin approach: use Firo's own estimatesmartfee(1) without the payout CPFP margin, floored at the relay minimum, instead of a hardcoded FIRO_MIN_FEE_RATE constant. On a quiet Firo node estimatesmartfee returns null (the normal state) and degrades to the relay floor; a genuine node/RPC error propagates so Firo fails closed (drops out of the fee cache) like Bitcoin rather than being advertised at 1. Removes the now-unused FIRO_MIN_FEE_RATE config and adds PayoutFiroService tests for the estimate/null/error branches. --- .../services/bitcoin-based-fee.service.ts | 2 +- .../payment-link-fee.service.spec.ts | 66 +++++++++++++++++++ .../services/payment-link-fee.service.ts | 19 +++++- .../__tests__/payout-firo.service.spec.ts | 24 +++++++ .../payout/services/payout-bitcoin.service.ts | 6 ++ .../payout/services/payout-firo.service.ts | 12 ++++ 6 files changed, 126 insertions(+), 3 deletions(-) create mode 100644 src/subdomains/core/payment-link/services/__tests__/payment-link-fee.service.spec.ts diff --git a/src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service.ts b/src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service.ts index 8bf81c9c79..a2fc89d9fd 100644 --- a/src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service.ts +++ b/src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service.ts @@ -17,7 +17,7 @@ export interface FeeConfig { } // Node's own minimum relay fee floor (sat/vB); broadcasts below this are rejected outright. -const MIN_FEE_RATE_SAT_VB = 1; +export const MIN_FEE_RATE_SAT_VB = 1; export abstract class BitcoinBasedFeeService { private readonly logger = new DfxLogger(BitcoinBasedFeeService); diff --git a/src/subdomains/core/payment-link/services/__tests__/payment-link-fee.service.spec.ts b/src/subdomains/core/payment-link/services/__tests__/payment-link-fee.service.spec.ts new file mode 100644 index 0000000000..9f2634deb8 --- /dev/null +++ b/src/subdomains/core/payment-link/services/__tests__/payment-link-fee.service.spec.ts @@ -0,0 +1,66 @@ +import { Blockchain } from 'src/integration/blockchain/shared/enums/blockchain.enum'; +import { BlockchainRegistryService } from 'src/integration/blockchain/shared/services/blockchain-registry.service'; +import { PayoutBitcoinService } from 'src/subdomains/supporting/payout/services/payout-bitcoin.service'; +import { PayoutFiroService } from 'src/subdomains/supporting/payout/services/payout-firo.service'; +import { PaymentLinkFeeService } from '../payment-link-fee.service'; + +describe('PaymentLinkFeeService', () => { + let service: PaymentLinkFeeService; + let blockchainRegistryService: jest.Mocked; + let payoutBitcoinService: jest.Mocked; + let payoutFiroService: jest.Mocked; + + beforeEach(() => { + blockchainRegistryService = {} as unknown as jest.Mocked; + + payoutBitcoinService = { + getCurrentFeeRate: jest.fn().mockResolvedValue(8), + getRecommendedFeeRate: jest.fn().mockResolvedValue(4), + } as unknown as jest.Mocked; + + payoutFiroService = { + getCurrentFeeRate: jest.fn().mockResolvedValue(6), + getRecommendedFeeRate: jest.fn().mockResolvedValue(3), + } as unknown as jest.Mocked; + + service = new PaymentLinkFeeService(blockchainRegistryService, payoutBitcoinService, payoutFiroService); + }); + + // --- calculateFee() Tests --- // + + describe('calculateFee()', () => { + it('should use the recommended rate (not the CPFP-multiplied payout rate) as the Firo customer minimum', async () => { + const fee = await service['calculateFee'](Blockchain.FIRO); + + expect(fee).toBe(3); + expect(payoutFiroService.getRecommendedFeeRate).toHaveBeenCalledTimes(1); + expect(payoutFiroService.getCurrentFeeRate).not.toHaveBeenCalled(); + }); + + it('should floor the Firo minimum at the relay minimum so protocol-fixed Spark payments pass', async () => { + // On a quiet Firo node estimatesmartfee yields the relay floor (~1 sat/vB); the customer + // minimum must never exceed what a Spark-spend to the transparent deposit address pays. + payoutFiroService.getRecommendedFeeRate.mockResolvedValueOnce(0.4); + + const fee = await service['calculateFee'](Blockchain.FIRO); + + expect(fee).toBe(1); + }); + + it('should use the recommended rate (not the CPFP-multiplied payout rate) as the Bitcoin customer minimum', async () => { + const fee = await service['calculateFee'](Blockchain.BITCOIN); + + expect(fee).toBe(4); + expect(payoutBitcoinService.getRecommendedFeeRate).toHaveBeenCalledTimes(1); + expect(payoutBitcoinService.getCurrentFeeRate).not.toHaveBeenCalled(); + }); + + it('should floor the Bitcoin minimum at the relay minimum when the recommended rate dips below it', async () => { + payoutBitcoinService.getRecommendedFeeRate.mockResolvedValueOnce(0.4); + + const fee = await service['calculateFee'](Blockchain.BITCOIN); + + expect(fee).toBe(1); + }); + }); +}); diff --git a/src/subdomains/core/payment-link/services/payment-link-fee.service.ts b/src/subdomains/core/payment-link/services/payment-link-fee.service.ts index a6a7f48cdf..ef2c786d66 100644 --- a/src/subdomains/core/payment-link/services/payment-link-fee.service.ts +++ b/src/subdomains/core/payment-link/services/payment-link-fee.service.ts @@ -1,6 +1,7 @@ import { Injectable, OnModuleInit } from '@nestjs/common'; import { CronExpression } from '@nestjs/schedule'; import { Environment, GetConfig } from 'src/config/config'; +import { MIN_FEE_RATE_SAT_VB } from 'src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service'; import { Blockchain } from 'src/integration/blockchain/shared/enums/blockchain.enum'; import { PaymentLinkBlockchains } from 'src/integration/blockchain/shared/util/blockchain.util'; import { DfxLogger } from 'src/shared/services/dfx-logger'; @@ -80,11 +81,25 @@ export class PaymentLinkFeeService implements OnModuleInit { return +(await client.getRecommendedGasPrice()); } + // The customer minimum is the network's own minimum for an inbound payment to confirm — it + // must NOT include the CPFP/default margin from getSendFeeRate, which exists only for DFX's + // own outbound spends. The value differs per chain because the chains do, but neither carries + // the payout margin. case Blockchain.BITCOIN: - return this.payoutBitcoinService.getCurrentFeeRate(); + // Bitcoin fees are user-adjustable and the chain can congest, so use the recommended + // (next-block) rate, which adapts to congestion — floored at the relay minimum so the + // advertised minimum is always relayable. + return Math.max(await this.payoutBitcoinService.getRecommendedFeeRate(), MIN_FEE_RATE_SAT_VB); case Blockchain.FIRO: - return this.payoutFiroService.getCurrentFeeRate(); + // Same principle as Bitcoin: Firo's own next-block rate without the payout margin, floored + // at the relay minimum so it stays relayable. The current OCP deposit address is transparent, + // so a Stack Wallet payment is a Spark-spend to it, whose fee sits at the relay floor and + // cannot be raised; Firo does not congest and its node usually returns no estimate, so this + // resolves to the relay floor in practice — exactly what that Spark-spend pays. A dedicated + // relay-floor cap belongs here only once a Spark `sm1…` deposit address is deployed, whose + // protocol-capped fee cannot follow a congestion-adaptive minimum. + return Math.max(await this.payoutFiroService.getRecommendedFeeRate(), MIN_FEE_RATE_SAT_VB); } } diff --git a/src/subdomains/supporting/payout/services/__tests__/payout-firo.service.spec.ts b/src/subdomains/supporting/payout/services/__tests__/payout-firo.service.spec.ts index 291bf0ff6d..ea59bacf36 100644 --- a/src/subdomains/supporting/payout/services/__tests__/payout-firo.service.spec.ts +++ b/src/subdomains/supporting/payout/services/__tests__/payout-firo.service.spec.ts @@ -32,6 +32,7 @@ describe('PayoutFiroService', () => { mintSpark: mintSparkSpy, getInfo: jest.fn(), getTx: jest.fn(), + estimateSmartFee: jest.fn(), } as unknown as jest.Mocked; const mockFiroService = { @@ -225,4 +226,27 @@ describe('PayoutFiroService', () => { expect(mockFeeService.getSendFeeRate).toHaveBeenCalledTimes(1); }); }); + + describe('getRecommendedFeeRate()', () => { + it('returns the node estimate without the payout margin', async () => { + (mockClient.estimateSmartFee as jest.Mock).mockResolvedValueOnce(3); + + await expect(service.getRecommendedFeeRate()).resolves.toBe(3); + expect(mockClient.estimateSmartFee).toHaveBeenCalledWith(1); + expect(mockFeeService.getSendFeeRate).not.toHaveBeenCalled(); + }); + + it('degrades to the relay floor when the quiet node returns no estimate (null)', async () => { + (mockClient.estimateSmartFee as jest.Mock).mockResolvedValueOnce(null); + + await expect(service.getRecommendedFeeRate()).resolves.toBe(1); + }); + + it('propagates a node/RPC error (fail-closed) instead of masking it as the relay floor', async () => { + const nodeError = new Error('Firo node unreachable'); + (mockClient.estimateSmartFee as jest.Mock).mockRejectedValueOnce(nodeError); + + await expect(service.getRecommendedFeeRate()).rejects.toBe(nodeError); + }); + }); }); diff --git a/src/subdomains/supporting/payout/services/payout-bitcoin.service.ts b/src/subdomains/supporting/payout/services/payout-bitcoin.service.ts index d1929a0708..080a8ca27a 100644 --- a/src/subdomains/supporting/payout/services/payout-bitcoin.service.ts +++ b/src/subdomains/supporting/payout/services/payout-bitcoin.service.ts @@ -63,6 +63,12 @@ export class PayoutBitcoinService extends PayoutBitcoinBasedService { return this.feeService.getSendFeeRate(); } + // Network's recommended (next-block) rate without the payout send margin (see getSendFeeRate). + // Used as the customer-facing minimum for inbound Open CryptoPay payments. + async getRecommendedFeeRate(): Promise { + return this.feeService.getRecommendedFeeRate(); + } + // Quantize each amount to 8 decimals before serializing to the RPC. Even though // BitcoinBasedStrategy.aggregatePayout already rounds once, downstream fee // adjustments and fixRoundingMismatch can re-introduce float artifacts. Reject diff --git a/src/subdomains/supporting/payout/services/payout-firo.service.ts b/src/subdomains/supporting/payout/services/payout-firo.service.ts index ba0ed01d62..9b7bd68556 100644 --- a/src/subdomains/supporting/payout/services/payout-firo.service.ts +++ b/src/subdomains/supporting/payout/services/payout-firo.service.ts @@ -1,4 +1,5 @@ import { Injectable } from '@nestjs/common'; +import { MIN_FEE_RATE_SAT_VB } from 'src/integration/blockchain/bitcoin/services/bitcoin-based-fee.service'; import { FiroClient } from 'src/integration/blockchain/firo/firo-client'; import { FiroFeeService } from 'src/integration/blockchain/firo/services/firo-fee.service'; import { FiroService } from 'src/integration/blockchain/firo/services/firo.service'; @@ -62,4 +63,15 @@ export class PayoutFiroService extends PayoutBitcoinBasedService { async getCurrentFeeRate(): Promise { return this.feeService.getSendFeeRate(); } + + // Network's recommended (next-block) rate without the payout send margin (see getCurrentFeeRate), + // used as the customer-facing minimum for inbound Open CryptoPay payments. Firo's estimatesmartfee + // returns null on a quiet node (little traffic) — the normal state, where the relay floor is the + // correct customer minimum (a Spark-spend to the transparent deposit address pays exactly that), so + // degrade to it. A genuine node/RPC error still propagates (estimateSmartFee returns null only when + // the node answers without an estimate; callNode rethrows connection errors), so a down node fails + // closed like Bitcoin — the chain drops out of the fee cache rather than being advertised at 1. + async getRecommendedFeeRate(): Promise { + return (await this.client.estimateSmartFee(1)) ?? MIN_FEE_RATE_SAT_VB; + } } From 7c4d8406bb16c5a615e154693ec75d58cbab62f3 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:20:45 +0200 Subject: [PATCH 2/5] fix(scrypt): fail the order on a rejected withdrawal (#4310) (#4327) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(scrypt): self-heal withdrawal completion after a missed WS event (#4310) A Scrypt withdrawal whose BalanceTransaction completion event is missed (a WS drop at the wrong moment) could never complete: getWithdrawalStatus read only the in-memory balance-transaction cache that the subscription fills, and getAllTransactions (the 5-minute EXCHANGE_TX_SYNC) read the same cache. A missed event therefore left the liquidity-management order polling forever and its in-flight amount double-counted in the balance snapshot until a full process restart. Give getWithdrawalStatus the same fresh-fetch fallback getOrderStatus already has — fetch from the Scrypt API on a cache miss, then cache the result — and make getAllTransactions fetch balance transactions fresh instead of reading the cache, reusing the existing connection.fetch(BALANCE_TRANSACTION) transport the constructor warm-up already uses. Either path now heals a missed event within minutes instead of never. Also reorder checkWithdrawCompletion so a FAILED/REJECTED status fails the order before the missing-txHash early return: a rejected withdrawal carries no txHash, so it previously returned false and polled forever instead of failing the order. * fix(scrypt): heal a missed withdrawal completion via the fresh 5-minute sync (#4310) Rework the self-heal after review. getWithdrawalStatus's cache-miss fetch could never fire: withdrawFunds() seeds a non-terminal balance-transaction cache entry via the permanent subscriber, so the cache is never empty for a stuck withdrawal, and a getOrderStatus-style cache-miss fallback is dead code for the exact "missed completion event" case this targets. Heal through the 5-minute EXCHANGE_TX_SYNC instead: getAllTransactions now fetches balance transactions fresh via fetchAll (full pagination, not a truncated single page), refreshes the cache conditionally so a non-terminal fetched record never overwrites a terminal cached one (a fetch cannot regress a completed withdrawal it raced against a live event), and on a fetch failure degrades loudly to the last-known-good cache instead of throwing (which would also discard the caller's concurrent trade fetch). getWithdrawalStatus reverts to a pure cache read; the FAILED/REJECTED reorder in the adapter is unchanged. * fix(scrypt): drop the fetch self-heal; keep only the rejected-withdrawal reorder (#4310) Adversarial review showed the fetch-based cache self-heal is unsound for balance transactions: the BalanceTransaction stream's schema forbids a StartDate filter (additionalProperties: false), so a since-filtered fetch is invalid and a wider window would still miss older stuck withdrawals; and fetch/fetchAll open an uncancelled server-side subscription, so running it per poll or every 5 minutes leaks streams. Revert scrypt.service.ts (getWithdrawalStatus and getAllTransactions) to their original cache-based form and drop the corresponding tests. This PR now carries only the completion-check reorder: evaluate FAILED/REJECTED before the missing-txHash early return, so a rejected withdrawal (which has no txHash) fails the order via OrderFailedException instead of polling forever. The heal for a missed completion event is handled correctly by a fetchAll catch-up on WS reconnect (unfiltered, as the constructor warm-up already does), which lands in the reconnect-resilience PR. * test(scrypt): use Object.assign entity and add a FAILED-without-txHash regression test (#4310) pr-ready review: build the order via Object.assign(new LiquidityManagementOrder(), ...) matching the repo convention, and add a dedicated FAILED-without-txHash case so both terminal statuses are pinned against the missing-txHash early return. --- .../actions/__tests__/scrypt.adapter.spec.ts | 132 ++++++++++++++++++ .../adapters/actions/scrypt.adapter.ts | 11 +- 2 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 src/subdomains/core/liquidity-management/adapters/actions/__tests__/scrypt.adapter.spec.ts diff --git a/src/subdomains/core/liquidity-management/adapters/actions/__tests__/scrypt.adapter.spec.ts b/src/subdomains/core/liquidity-management/adapters/actions/__tests__/scrypt.adapter.spec.ts new file mode 100644 index 0000000000..c1997ddf51 --- /dev/null +++ b/src/subdomains/core/liquidity-management/adapters/actions/__tests__/scrypt.adapter.spec.ts @@ -0,0 +1,132 @@ +import { createMock } from '@golevelup/ts-jest'; +import { Blockchain } from 'src/integration/blockchain/shared/enums/blockchain.enum'; +import { ScryptTransactionStatus, ScryptWithdrawStatus } from 'src/integration/exchange/dto/scrypt.dto'; +import { ScryptService } from 'src/integration/exchange/services/scrypt.service'; +import { AssetService } from 'src/shared/models/asset/asset.service'; +import { DexService } from 'src/subdomains/supporting/dex/services/dex.service'; +import { PricingService } from 'src/subdomains/supporting/pricing/services/pricing.service'; +import { LiquidityManagementOrder } from '../../../entities/liquidity-management-order.entity'; +import { OrderFailedException } from '../../../exceptions/order-failed.exception'; +import { LiquidityManagementOrderRepository } from '../../../repositories/liquidity-management-order.repository'; +import { ScryptAdapter, ScryptAdapterCommands } from '../scrypt.adapter'; + +const DEST_ENV = 'TEST_SCRYPT_WITHDRAW_ADDR'; + +function createWithdrawOrder(overrides: Partial = {}): LiquidityManagementOrder { + return Object.assign(new LiquidityManagementOrder(), { + correlationId: 'corr-1', + action: { + command: ScryptAdapterCommands.WITHDRAW, + paramMap: { + destinationAddress: DEST_ENV, + destinationBlockchain: Blockchain.ETHEREUM, + }, + }, + outputAmount: undefined, + ...overrides, + }); +} + +describe('ScryptAdapter', () => { + let adapter: ScryptAdapter; + let scryptService: ScryptService; + let dexService: DexService; + let orderRepo: LiquidityManagementOrderRepository; + let pricingService: PricingService; + let assetService: AssetService; + + beforeEach(() => { + process.env[DEST_ENV] = '0xabc'; + + scryptService = createMock({ name: 'Scrypt' }); + dexService = createMock(); + orderRepo = createMock(); + pricingService = createMock(); + assetService = createMock(); + + adapter = new ScryptAdapter(scryptService, dexService, orderRepo, pricingService, assetService); + }); + + afterEach(() => { + delete process.env[DEST_ENV]; + jest.restoreAllMocks(); + }); + + describe('checkWithdrawCompletion', () => { + it('throws OrderFailedException when status is REJECTED with no txHash', async () => { + const withdrawal: ScryptWithdrawStatus = { + id: 'w-1', + status: ScryptTransactionStatus.REJECTED, + rejectReason: 'InvalidAddress', + rejectText: 'bad address', + }; + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValue(withdrawal); + + await expect(adapter.checkCompletion(createWithdrawOrder())).rejects.toThrow(OrderFailedException); + await expect(adapter.checkCompletion(createWithdrawOrder())).rejects.toThrow( + /Withdrawal corr-1 has failed with status Rejected/, + ); + expect(dexService.checkTransferCompletion).not.toHaveBeenCalled(); + }); + + it('throws OrderFailedException when status is FAILED', async () => { + const withdrawal: ScryptWithdrawStatus = { + id: 'w-2', + status: ScryptTransactionStatus.FAILED, + txHash: '0xdead', + rejectReason: 'NetworkError', + rejectText: 'timeout', + }; + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValue(withdrawal); + + await expect(adapter.checkCompletion(createWithdrawOrder())).rejects.toThrow(OrderFailedException); + expect(dexService.checkTransferCompletion).not.toHaveBeenCalled(); + }); + + it('throws OrderFailedException when status is FAILED with no txHash', async () => { + const withdrawal: ScryptWithdrawStatus = { + id: 'w-3', + status: ScryptTransactionStatus.FAILED, + rejectReason: 'NetworkError', + rejectText: 'timeout', + }; + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValue(withdrawal); + + await expect(adapter.checkCompletion(createWithdrawOrder())).rejects.toThrow(OrderFailedException); + expect(dexService.checkTransferCompletion).not.toHaveBeenCalled(); + }); + + it('returns false when withdrawal is missing or still pending without a txHash', async () => { + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValueOnce(null); + + await expect(adapter.checkCompletion(createWithdrawOrder())).resolves.toBe(false); + + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValueOnce({ + id: 'w-pending', + status: ScryptTransactionStatus.COMPLETED, + // no txHash yet + }); + + await expect(adapter.checkCompletion(createWithdrawOrder())).resolves.toBe(false); + expect(dexService.checkTransferCompletion).not.toHaveBeenCalled(); + }); + + it('sets order.outputAmount and delegates to dexService when withdrawal has a txHash', async () => { + const withdrawal: ScryptWithdrawStatus = { + id: 'w-ok', + status: ScryptTransactionStatus.COMPLETED, + txHash: '0xsuccess', + amount: 3.25, + }; + jest.spyOn(scryptService, 'getWithdrawalStatus').mockResolvedValue(withdrawal); + jest.spyOn(dexService, 'checkTransferCompletion').mockResolvedValue(true); + + const order = createWithdrawOrder(); + const result = await adapter.checkCompletion(order); + + expect(result).toBe(true); + expect(order.outputAmount).toBe(3.25); + expect(dexService.checkTransferCompletion).toHaveBeenCalledWith('0xsuccess', Blockchain.ETHEREUM); + }); + }); +}); diff --git a/src/subdomains/core/liquidity-management/adapters/actions/scrypt.adapter.ts b/src/subdomains/core/liquidity-management/adapters/actions/scrypt.adapter.ts index 12032dee0c..f60dc692f1 100644 --- a/src/subdomains/core/liquidity-management/adapters/actions/scrypt.adapter.ts +++ b/src/subdomains/core/liquidity-management/adapters/actions/scrypt.adapter.ts @@ -189,10 +189,8 @@ export class ScryptAdapter extends LiquidityActionAdapter { const { correlationId } = order; const withdrawal = await this.scryptService.getWithdrawalStatus(correlationId); - if (!withdrawal?.txHash) { - this.logger.verbose(`No withdrawal id for id ${correlationId} at ${this.scryptService.name} found`); - return false; - } else if ([ScryptTransactionStatus.FAILED, ScryptTransactionStatus.REJECTED].includes(withdrawal.status)) { + + if (withdrawal && [ScryptTransactionStatus.FAILED, ScryptTransactionStatus.REJECTED].includes(withdrawal.status)) { const rejectMessage = withdrawal.rejectReason ? `${withdrawal.rejectReason} (${withdrawal.rejectText})` : 'unknown reason'; @@ -201,6 +199,11 @@ export class ScryptAdapter extends LiquidityActionAdapter { ); } + if (!withdrawal?.txHash) { + this.logger.verbose(`No withdrawal id for id ${correlationId} at ${this.scryptService.name} found`); + return false; + } + order.outputAmount = withdrawal.amount; const { blockchain } = this.parseWithdrawParams(order.action.paramMap); From 428c815b31c2f1df024ab2b6fb29f8f55794279c Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:45:47 +0200 Subject: [PATCH 3/5] fix(log): clamp the unfiltered pending legs to match the filtered ones (#4310) (#4338) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(log): clamp the unfiltered pending legs to match the filtered ones (#4310) The filtered pending legs (fromKraken/toKraken/fromScrypt/toScrypt) are each clamped to 0 when negative, but their unfiltered counterparts were not. When useUnfilteredTx is true, an unclamped negative unfiltered leg (observed: asset 405 approx -5.0M) drove totalPlusPending negative and fired the 'totalPlusPending < 0' verbose log every minute. Clamp the four unfiltered legs symmetrically, placed after the filtered/unfiltered discrepancy comparisons so those still compare raw values. The root cause of a persistently-negative unfiltered leg is a separate follow-up. * fix(log): log the unfiltered clamp components and cover the useUnfilteredTx path (#4310) pr-ready review: the four new unfiltered-leg <0 verbose logs now include their constituent pending-amount breakdown (matching the filtered siblings), which the separate follow-up into the persistently-negative unfiltered leg will need; and add a test exercising the useUnfilteredTx=true clamp path (previously entirely untested). * test(log): isolate the per-leg unfiltered-toKraken clamp from the aggregate clamp (#4310) pr-ready round 2: the negative-leg clamp test was vacuous — the downstream aggregate totalPlusPending<0 clamp floors the same scenario, so reverting the per-leg clamp still left it green. Assert (via the verbose log) that the per-leg 'toKrakenUnfiltered balance < 0' path fires and the aggregate 'totalPlusPending < 0' path does NOT, so the test now fails if the per-leg clamp is removed. --- .../log/__tests__/log-job.service.spec.ts | 92 ++++++++++++++++++- .../supporting/log/log-job.service.ts | 45 ++++++++- 2 files changed, 132 insertions(+), 5 deletions(-) diff --git a/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts b/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts index d8c3e4a560..bde4820a3d 100644 --- a/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts +++ b/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts @@ -33,7 +33,7 @@ import { createCustomBankTx } from '../../bank-tx/bank-tx/__mocks__/bank-tx.enti import { BankTx, BankTxIndicator, BankTxType } from '../../bank-tx/bank-tx/entities/bank-tx.entity'; import { Bank } from '../../bank/bank/bank.entity'; import { BankService } from '../../bank/bank/bank.service'; -import { frickCHF, frickEUR, olkyEUR, yapealCHF } from '../../bank/bank/__mocks__/bank.entity.mock'; +import { frickCHF, frickEUR, olkyEUR, yapealCHF, yapealEUR } from '../../bank/bank/__mocks__/bank.entity.mock'; import { IbanBankName } from '../../bank/bank/dto/bank.dto'; import { createCustomFiatOutput } from '../../fiat-output/__mocks__/fiat-output.entity.mock'; import { PayInService } from '../../payin/services/payin.service'; @@ -1830,4 +1830,94 @@ describe('LogJobService', () => { expect(assetLog[yapealAsset.id].plusBalance.pending?.toScrypt ?? 0).toBe(0); }); }); + + describe('useUnfilteredTx per-leg clamp (toKrakenUnfiltered < 0)', () => { + beforeEach(() => { + (BankService as unknown as { ibanCache: Map }).ibanCache.clear(); + (BankService as unknown as { ibanCache: Map }).ibanCache.set( + `${IbanBankName.YAPEAL}-CHF`, + yapealCHF.iban, + ); + }); + + afterEach(() => { + (BankService as unknown as { ibanCache: Map }).ibanCache.clear(); + }); + + function setupUnfilteredToKrakenClamp(depositTx: ReturnType) { + jest.spyOn(settingService, 'getCustomBalanceSettings').mockResolvedValue({ assets: [], addresses: [] }); + jest.spyOn(settingService, 'getObj').mockImplementation(async (key, defaultValue) => { + if (key === 'financeLogUnfilteredTx') return true as never; + if (key === 'financeLogPairIds') + return { + fromKraken: { chf: { bankTxId: 0, exchangeTxId: 0 }, eur: { bankTxId: 0, exchangeTxId: 0 } }, + toKraken: { chf: { bankTxId: 0, exchangeTxId: 0 }, eur: { bankTxId: 0, exchangeTxId: 0 } }, + toScrypt: { chf: { bankTxId: 0, exchangeTxId: 0 }, eur: { bankTxId: 0, exchangeTxId: 0 } }, + } as never; + return defaultValue as never; + }); + + jest.spyOn(bankService, 'getBankInternal').mockImplementation(async (name, currency) => { + if (name === IbanBankName.YAPEAL && currency === 'CHF') return yapealCHF; + if (name === IbanBankName.YAPEAL && currency === 'EUR') return yapealEUR; + if (name === IbanBankName.FRICK && currency === 'CHF') return frickCHF; + if (name === IbanBankName.FRICK && currency === 'EUR') return frickEUR; + if (name === IbanBankName.OLKY && currency === 'EUR') return olkyEUR; + return Object.assign(new Bank(), { name, currency, iban: `IBAN_${name}_${currency}`, bic: 'BICTEST' }); + }); + + jest.spyOn(liquidityManagementPipelineService, 'getPendingTx').mockResolvedValue([]); + jest.spyOn(payInService, 'getPendingPayIns').mockResolvedValue([]); + jest.spyOn(buyFiatService, 'getPendingTransactions').mockResolvedValue([]); + jest.spyOn(buyCryptoService, 'getPendingTransactions').mockResolvedValue([]); + jest.spyOn(bankTxService, 'getPendingTx').mockResolvedValue([]); + jest.spyOn(bankTxRepeatService, 'getPendingTx').mockResolvedValue([]); + jest.spyOn(bankTxReturnService, 'getPendingTx').mockResolvedValue([]); + jest.spyOn(bankTxService, 'getRecentBankToBankTx').mockResolvedValue([]); + jest.spyOn(payoutService, 'getRecentPayoutSentCorrelationIds').mockResolvedValue(new Set()); + jest.spyOn(paymentBalanceService, 'getPaymentBalances').mockResolvedValue(new Map()); + jest.spyOn(bankTxService, 'getRecentExchangeTx').mockResolvedValue([]); + jest + .spyOn(exchangeTxService, 'getRecentExchangeTx') + .mockImplementation(async (_minId, exchange, _types) => (exchange === ExchangeName.KRAKEN ? [depositTx] : [])); + } + + it('floors a negative unfiltered toKraken leg so plusBalance.total is 0 (not -10000)', async () => { + const yapealChfAsset = createCustomAsset({ + id: 8003, + blockchain: Blockchain.YAPEAL, + dexName: 'CHF', + sellable: true, + }); + + // Unmatched Kraken DEPOSIT credited to Yapeal CHF BIC → pendingBankAmount = -amount for toKrakenUnfiltered + const theDepositTx = createCustomExchangeTx({ + id: 5001, + type: ExchangeTxType.DEPOSIT, + status: 'ok', + currency: 'CHF', + method: 'Bank Frick (SIC) International', + address: yapealCHF.bic.padEnd(11, 'XXX'), + amount: 10000, + }); + setupUnfilteredToKrakenClamp(theDepositTx); + + const verboseSpy = jest.spyOn(service['logger'], 'verbose'); + + const assetLog = await service['getAssetLog']([yapealChfAsset]); + + // Without the per-leg clamp, toKrakenUnfiltered (-10000) would make plusBalance.total negative + expect(assetLog[yapealChfAsset.id].plusBalance.total).toBe(0); + // pending is only populated when totalPlusPending !== 0; after floor both read as 0 + expect(assetLog[yapealChfAsset.id].plusBalance.pending?.toKraken ?? 0).toBe(0); + + // Prove the PER-LEG clamp fired (not only the aggregate totalPlusPending clamp downstream): + // with per-leg active, totalPlusPending is already 0 so the aggregate clamp never logs. + // If per-leg flooring is reverted, toKrakenUnfiltered stays negative → aggregate clamp logs instead. + expect(verboseSpy.mock.calls.some((call) => String(call[0]).includes('toKrakenUnfiltered balance < 0'))).toBe( + true, + ); + expect(verboseSpy.mock.calls.some((call) => String(call[0]).includes('totalPlusPending < 0'))).toBe(false); + }); + }); }); diff --git a/src/subdomains/supporting/log/log-job.service.ts b/src/subdomains/supporting/log/log-job.service.ts index 77867dddd8..2cd855c8fd 100644 --- a/src/subdomains/supporting/log/log-job.service.ts +++ b/src/subdomains/supporting/log/log-job.service.ts @@ -808,11 +808,11 @@ export class LogJobService { : 0; const pendingScryptBankMinusAmountUnfiltered = 0; - const fromKrakenUnfiltered = + let fromKrakenUnfiltered = pendingChfKrakenYapealPlusAmountUnfiltered + pendingEurKrakenYapealPlusAmountUnfiltered + pendingKrakenYapealMinusAmountUnfiltered; - const toKrakenUnfiltered = + let toKrakenUnfiltered = pendingYapealKrakenPlusAmountUnfiltered + pendingChfYapealKrakenMinusAmountUnfiltered + pendingEurYapealKrakenMinusAmountUnfiltered; @@ -825,11 +825,11 @@ export class LogJobService { let fromScrypt = pendingChfScryptBankPlusAmount + pendingEurScryptBankPlusAmount + pendingScryptBankMinusAmount; let toScrypt = pendingBankScryptPlusAmount + pendingChfBankScryptMinusAmount + pendingEurBankScryptMinusAmount; - const fromScryptUnfiltered = + let fromScryptUnfiltered = pendingChfScryptBankPlusAmountUnfiltered + pendingEurScryptBankPlusAmountUnfiltered + pendingScryptBankMinusAmountUnfiltered; - const toScryptUnfiltered = + let toScryptUnfiltered = pendingBankScryptPlusAmountUnfiltered + pendingChfBankScryptMinusAmountUnfiltered + pendingEurBankScryptMinusAmountUnfiltered; @@ -901,6 +901,43 @@ export class LogJobService { fromScrypt = 0; } + if (fromKrakenUnfiltered < 0) { + errors.push(`fromKrakenUnfiltered < 0`); + this.logger.verbose( + `Error in financial log, fromKrakenUnfiltered balance < 0 for asset: ${curr.id}, pendingChfPlusAmount: + ${pendingChfKrakenYapealPlusAmountUnfiltered}, pendingEurPlusAmount: ${pendingEurKrakenYapealPlusAmountUnfiltered}, + pendingMinusAmount: ${pendingKrakenYapealMinusAmountUnfiltered}`, + ); + fromKrakenUnfiltered = 0; + } + if (toKrakenUnfiltered < 0) { + errors.push(`toKrakenUnfiltered < 0`); + this.logger.verbose( + `Error in financial log, toKrakenUnfiltered balance < 0 for asset: ${curr.id}, pendingPlusAmount: + ${pendingYapealKrakenPlusAmountUnfiltered}, pendingChfMinusAmount: ${pendingChfYapealKrakenMinusAmountUnfiltered}, + pendingEurMinusAmount: ${pendingEurYapealKrakenMinusAmountUnfiltered}`, + ); + toKrakenUnfiltered = 0; + } + if (fromScryptUnfiltered < 0) { + errors.push(`fromScryptUnfiltered < 0`); + this.logger.verbose( + `Error in financial log, fromScryptUnfiltered balance < 0 for asset: ${curr.id}, pendingChfPlusAmount: + ${pendingChfScryptBankPlusAmountUnfiltered}, pendingEurPlusAmount: ${pendingEurScryptBankPlusAmountUnfiltered}, + pendingMinusAmount: ${pendingScryptBankMinusAmountUnfiltered}`, + ); + fromScryptUnfiltered = 0; + } + if (toScryptUnfiltered < 0) { + errors.push(`toScryptUnfiltered < 0`); + this.logger.verbose( + `Error in financial log, toScryptUnfiltered balance < 0 for asset: ${curr.id}, pendingPlusAmount: + ${pendingBankScryptPlusAmountUnfiltered}, pendingChfMinusAmount: ${pendingChfBankScryptMinusAmountUnfiltered}, + pendingEurMinusAmount: ${pendingEurBankScryptMinusAmountUnfiltered}`, + ); + toScryptUnfiltered = 0; + } + // total pending balance let totalPlusPending = cryptoInput + From 70774e2a4813914aa588183ea3918866b0eadcf7 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:46:06 +0200 Subject: [PATCH 4/5] fix(scrypt): resubscribe on every WebSocket reconnect with bounded backoff (#4310) (#4331) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(scrypt): resubscribe on every WebSocket reconnect and retry with bounded backoff (#4310) The Scrypt WS reconnect had two gaps that let a dropped connection silently stop delivering events for the rest of the process lifetime: - The scheduled reconnect was single-shot: one setTimeout -> connect(), whose .catch only logged. After a single failed reconnect (e.g. a 401) nothing ever retried. - resubscribeToStreams ran only from that dead single-shot path. Any implicit reconnect via ensureConnected() (a business call such as a price fetch) rebuilt the raw socket without restoring subscriptions, so the socket healed while the BalanceTransaction stream stayed unsubscribed. Move resubscription into connect()'s success path so every re-connect restores the streams, guarded by hasEverConnected so the first connect (whose subscriptions are sent directly by the initial subscribe() calls) does not double-send. Replace the single-shot reconnect with a bounded-backoff loop (5s doubling, capped at 60s) that retries indefinitely and logs reconnect success, guarded by isReconnecting against overlapping loops. resubscribeToStreams no longer clears activeStreams up front, so a transient per-stream subscribe failure leaves the stream queued for the next reconnect instead of dropping it permanently. An explicit fetchAll catch-up on reconnect (to recover events missed during the outage) and the fetch/fetchAll subscription-cancel fix follow in a separate PR. * fix(scrypt): share reconnect readiness and fail loud on a mid-resubscribe drop (#4310) Rework the reconnect after review found two concurrency blockers: - A caller joining an in-flight reconnect (connect()'s CONNECTING branch) received the raw handshake promise, so it could send a money-path request as soon as the socket opened — before the streams were resubscribed, risking a missed confirmation push. connectionPromise now covers full readiness via establishConnection (handshake -> revalidate -> resubscribe on a re-connect -> revalidate -> CONNECTED); connectionState stays CONNECTING until the end, so a joiner waits for resubscription instead of proceeding on a half-ready socket. - resubscribeToStreams swallowed all errors and connect() never revalidated the socket after it, so a drop mid-resubscribe resolved connect() as a false "reconnected" and left the connection dead with no retry. establishConnection now asserts the socket is open after resubscription, so a mid-resubscribe drop rejects and the backoff loop retries. resubscribeToStreams sends on the current socket directly (sendSubscriptionOnSocket, no ensureConnected) so it cannot recurse into connect(). Also: a 15s handshake timeout so a black-hole connect cannot wedge the loop; full jitter on the backoff with an escalation log every 10 attempts; and disconnect() now cancels a pending reconnect timer. * fix(scrypt): close a mid-establish reconnect race that could spawn duplicate sockets (#4310) A socket drop DURING establishConnection (state CONNECTING) previously flipped connectionState to DISCONNECTED, letting a concurrent connect() start a second establishConnection while the first was still in flight; the stale first attempt's rejection then clobbered the second's freshly-installed state, spawning duplicate sockets and inconsistent connection state. handleDisconnection now only resets connectionState when the drop hit an already CONNECTED socket, so a mid-establish close leaves state CONNECTING and concurrent callers keep joining the in-flight connectionPromise instead of starting a second establish. connect()'s catch additionally only resets state if it still owns the in-flight promise, so a stale establishConnection's rejection can never clobber a newer attempt. Strengthen the drop-mid-resubscribe test to emit a real close event (so the real handleDisconnection runs) and add a concurrency test asserting no second establishConnection is started; both were mutation-checked (they fail when the fix is reverted). * fix(scrypt): tie reconnect attempts to a generation token so stale socket events cannot corrupt a newer connection (#4310) A review found a residual class of races: a socket event (open/close) or an establishConnection step from an abandoned connection attempt could still mutate the shared connection state that a newer attempt (or disconnect()) had moved on to — e.g. a superseded socket's delayed close nulling a newer healthy this.ws and scheduling a bogus reconnect, or disconnect() failing to cancel an in-flight attempt or its backoff. Give every connection attempt a generation (connectionGeneration): connect() mints a new one, the socket's open/close listeners and establishConnection capture it, and they only mutate shared state while it is still current — a superseded socket's open terminates instead of being adopted, its close is ignored, and establishConnection aborts. disconnect() bumps the generation to supersede any in-flight attempt, and the backoff loop short-circuits on isReconnecting. This closes the class at its root (a missing attempt identity) instead of patching individual interleavings. These were only reachable via disconnect() (no production callers today) or an extremely narrow handshake-timeout timing, so this is latent hardening — but it makes the state machine correct if disconnect() is ever wired into a shutdown path. * fix(scrypt): reset connection state in disconnect() before the pre-open early return (#4310) disconnect() bumped the generation and cleared the reconnect timer, but during a pre-open handshake (this.ws still undefined) it early-returned without resetting connectionState/connectionPromise — so a later connect() joined the superseded, doomed promise via the single-flight guard instead of starting a fresh attempt. Reset connectionState = DISCONNECTED and connectionPromise = undefined unconditionally, before the `if (!this.ws) return`. * fix(scrypt): address pr-ready review — log the error object, full disconnect cleanup, doc/naming (#4310) - scheduleReconnect passes the error as the logger's second argument so the reconnect-failure class keeps its stack trace and span.recordException. - disconnect() clears pendingRequests/subscriptions/activeStreams unconditionally (only ws.close() is gated), so a pre-open-handshake disconnect fully resets state. - correct the establishConnection comment (close-before-open is bounded by the handshake timeout, not a direct reject); drop the now-unnecessary Array.from in resubscribeToStreams; rename 'full jitter' -> 'equal jitter'; document the latent subscribe-while-disconnected double-send on subscribeToStream. * fix(exchange): scope scrypt reconnect loop to an epoch so a superseded attempt cannot disturb a newer loop (#4310) pr-ready round 2: the reconnect loop guarded its timer/then/catch continuations on the bare isReconnecting flag, so a disconnect() that supersedes an in-flight attempt and is followed by a fresh drop+loop could let the stale attempt's late continuation clear the live loop's isReconnecting or overwrite its reconnectTimer. Scope every scheduleReconnect continuation to a reconnectEpoch (bumped on disconnect and at each new loop start) so a superseded attempt no-ops. Also reword the subscribeToStream JSDoc to the real already-active-stream invariant and correct a stale flushPromises test comment. * fix(exchange): clear reconnect state on full connect and reset reuse flag on disconnect (#4310) pr-ready round 3: two confirm-review findings. (1) isReconnecting/reconnectTimer were reset only in scheduleReconnect's own success continuation, so a business call that healed the socket via ensureConnected()->connect() during the backoff window left them stale — clear both once establishConnection reaches CONNECTED, whichever path connected. (2) disconnect() did not reset hasEverConnected, so reusing the connection (disconnect then subscribeToStream) double-sent the SUBSCRIBE frame — reset it for a clean full-reset. Both pinned by mutation-checked tests. --- .../scrypt-websocket-connection.spec.ts | 763 ++++++++++++++++++ .../services/scrypt-websocket-connection.ts | 189 +++-- 2 files changed, 904 insertions(+), 48 deletions(-) create mode 100644 src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts diff --git a/src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts b/src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts new file mode 100644 index 0000000000..24c8d4de1f --- /dev/null +++ b/src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts @@ -0,0 +1,763 @@ +import { EventEmitter as MockEventEmitter } from 'events'; +import Ws from 'ws'; +import { ScryptMessageType, ScryptWebSocketConnection } from '../scrypt-websocket-connection'; + +type MockWebSocketInstance = MockEventEmitter & { + url: string; + options?: unknown; + readyState: number; + send: jest.Mock; + close: jest.Mock; + terminate: jest.Mock; + open: () => void; + fail: (error?: Error) => void; + remoteClose: (code?: number, reason?: string) => void; +}; + +type MockWebSocketConstructor = { + new (url: string, options?: unknown): MockWebSocketInstance; + OPEN: number; + CONNECTING: number; + CLOSING: number; + CLOSED: number; + instances: MockWebSocketInstance[]; +}; + +jest.mock('ws', () => { + class MockWebSocket extends MockEventEmitter { + static OPEN = 1; + static CONNECTING = 0; + static CLOSING = 2; + static CLOSED = 3; + static instances: MockWebSocket[] = []; + + readyState = MockWebSocket.CONNECTING; + send = jest.fn(); + close = jest.fn(() => { + this.readyState = MockWebSocket.CLOSED; + }); + terminate = jest.fn(() => { + this.readyState = MockWebSocket.CLOSED; + }); + url: string; + options?: unknown; + + constructor(url: string, options?: unknown) { + super(); + this.url = url; + this.options = options; + MockWebSocket.instances.push(this); + } + + open(): void { + this.readyState = MockWebSocket.OPEN; + this.emit('open'); + } + + fail(error: Error = new Error('connect failed')): void { + this.emit('error', error); + } + + remoteClose(code = 1006, reason = 'abnormal'): void { + this.readyState = MockWebSocket.CLOSED; + this.emit('close', code, reason); + } + } + + return MockWebSocket; +}); + +// jest.mock is hoisted above imports; this binding receives the mock constructor. +const WebSocket = Ws as unknown as MockWebSocketConstructor; + +async function flushPromises(): Promise { + // Drain microtasks from connect() → establishConnection → resubscribeToStreams → + // sendSubscriptionOnSocket (and nested promise chains) under fake timers. A fixed small number of + // rounds is not reliably enough for the deepest chains, so loop generously. + for (let i = 0; i < 30; i++) { + await Promise.resolve(); + } +} + +function latestWs(): MockWebSocketInstance { + const { instances } = WebSocket; + if (instances.length === 0) throw new Error('No MockWebSocket instances created'); + return instances[instances.length - 1]; +} + +function subscribeMessages(ws: MockWebSocketInstance): unknown[] { + return ws.send.mock.calls.map(([payload]) => JSON.parse(payload as string)).filter((msg) => msg.type === 'subscribe'); +} + +/** Max delay for a given reconnect attempt (jitter is applied on top: capped/2 .. capped). */ +function maxDelayForAttempt(attempt: number): number { + return Math.min(5000 * 2 ** attempt, 60000); +} + +describe('ScryptWebSocketConnection', () => { + let connection: ScryptWebSocketConnection; + let loggerInfo: jest.SpyInstance; + let loggerWarn: jest.SpyInstance; + let loggerError: jest.SpyInstance; + + beforeEach(() => { + jest.useFakeTimers(); + WebSocket.instances = []; + connection = new ScryptWebSocketConnection('wss://scrypt.example/ws', 'api-key', 'api-secret'); + + loggerInfo = jest.spyOn((connection as any).logger, 'info').mockImplementation(() => undefined); + loggerWarn = jest.spyOn((connection as any).logger, 'warn').mockImplementation(() => undefined); + loggerError = jest.spyOn((connection as any).logger, 'error').mockImplementation(() => undefined); + }); + + afterEach(() => { + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + async function firstConnectWithStream( + streamName: ScryptMessageType = ScryptMessageType.BALANCE_TRANSACTION, + ): Promise { + connection.subscribeToStream(streamName, () => undefined); + const ws = latestWs(); + ws.open(); + await flushPromises(); + return ws; + } + + async function firstConnectWithStreams(streamNames: ScryptMessageType[]): Promise { + for (const streamName of streamNames) { + connection.subscribeToStream(streamName, () => undefined); + } + const ws = latestWs(); + ws.open(); + await flushPromises(); + return ws; + } + + /** Advance past the (jittered) reconnect timer for the given attempt and flush microtasks. */ + async function fireReconnectAttempt(attempt: number): Promise { + jest.advanceTimersByTime(maxDelayForAttempt(attempt)); + await flushPromises(); + } + + it('resubscribes active streams on reconnect after a close', async () => { + const resubscribeSpy = jest.spyOn(connection as any, 'resubscribeToStreams'); + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + + expect(subscribeMessages(firstWs)).toHaveLength(1); + expect(resubscribeSpy).not.toHaveBeenCalled(); + expect((connection as any).hasEverConnected).toBe(true); + + // Simulate unexpected disconnect while CONNECTED → backoff reconnect loop. + firstWs.remoteClose(1006, 'gone'); + expect((connection as any).isReconnecting).toBe(true); + + const constructCountBefore = WebSocket.instances.length; + await fireReconnectAttempt(0); + + expect(WebSocket.instances.length).toBe(constructCountBefore + 1); + const reconnectedWs = latestWs(); + reconnectedWs.open(); + await flushPromises(); + + expect(resubscribeSpy).toHaveBeenCalledTimes(1); + expect(subscribeMessages(reconnectedWs)).toHaveLength(1); + expect(subscribeMessages(reconnectedWs)[0]).toEqual( + expect.objectContaining({ + type: 'subscribe', + streams: [expect.objectContaining({ name: ScryptMessageType.BALANCE_TRANSACTION })], + }), + ); + expect((connection as any).isReconnecting).toBe(false); + expect(loggerInfo).toHaveBeenCalledWith(expect.stringMatching(/reconnected \(after 1 attempt/)); + }); + + it('does not double-subscribe on the first connect', async () => { + const resubscribeSpy = jest.spyOn(connection as any, 'resubscribeToStreams'); + + connection.subscribeToStream(ScryptMessageType.BALANCE, () => undefined); + const ws = latestWs(); + ws.open(); + await flushPromises(); + + expect(resubscribeSpy).not.toHaveBeenCalled(); + expect(subscribeMessages(ws)).toHaveLength(1); + expect(subscribeMessages(ws)[0]).toEqual( + expect.objectContaining({ + type: 'subscribe', + streams: [expect.objectContaining({ name: ScryptMessageType.BALANCE })], + }), + ); + expect((connection as any).hasEverConnected).toBe(true); + }); + + it('retries reconnect with bounded exponential backoff until success', async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(); + + firstWs.remoteClose(1006, 'drop'); + expect(scheduleSpy).toHaveBeenCalledWith(0, expect.any(Number)); + expect((connection as any).isReconnecting).toBe(true); + const loopEpoch = scheduleSpy.mock.calls[0][1] as number; + + // attempt 0 → delay in [2500, 5000] + await fireReconnectAttempt(0); + const attempt0 = latestWs(); + const authError = new Error('401 unauthorized'); + attempt0.fail(authError); + // close after error while CONNECTING: wasConnected=false, no extra loop + attempt0.remoteClose(1006, 'auth fail'); + await flushPromises(); + + // connectWebSocket's ws.on('error') logs via logger.error before rejecting + expect(loggerError).toHaveBeenCalledWith('Scrypt WebSocket error:', authError); + expect(loggerWarn).toHaveBeenCalledWith(expect.stringMatching(/reconnect attempt 1 failed/), expect.any(Error)); + expect(scheduleSpy).toHaveBeenCalledWith(1, loopEpoch); + + // attempt 1 → delay in [5000, 10000] + await fireReconnectAttempt(1); + const attempt1 = latestWs(); + attempt1.fail(new Error('still down')); + attempt1.remoteClose(); + await flushPromises(); + + expect(loggerWarn).toHaveBeenCalledWith(expect.stringMatching(/reconnect attempt 2 failed/), expect.any(Error)); + expect(scheduleSpy).toHaveBeenCalledWith(2, loopEpoch); + + // attempt 2 → delay in [10000, 20000], then succeed + await fireReconnectAttempt(2); + const attempt2 = latestWs(); + attempt2.open(); + await flushPromises(); + + expect(loggerInfo).toHaveBeenCalledWith(expect.stringMatching(/reconnected \(after 3 attempt/)); + expect((connection as any).isReconnecting).toBe(false); + + // attempts scheduled: 0, 1, 2 (capped bases 5s/10s/20s; actual delay is jittered in [capped/2, capped]) + // Same epoch throughout the loop — retries must not bump reconnectEpoch. + expect(scheduleSpy.mock.calls.map(([attempt, epoch]: [number, number]) => [attempt, epoch])).toEqual([ + [0, loopEpoch], + [1, loopEpoch], + [2, loopEpoch], + ]); + }); + + it('resubscribes when an implicit reconnect is driven by a business call (ensureConnected)', async () => { + const resubscribeSpy = jest.spyOn(connection as any, 'resubscribeToStreams'); + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + expect(resubscribeSpy).not.toHaveBeenCalled(); + + // Tear down the live socket (starts the reconnect loop with a delayed timer). + firstWs.remoteClose(1006, 'gone'); + expect((connection as any).connectionState).toBe('disconnected'); + expect((connection as any).hasEverConnected).toBe(true); + + // Before the backoff timer fires, a business call must heal + resubscribe via ensureConnected. + const sendPromise = connection.send(ScryptMessageType.TRADE, [{ side: 'buy' }]); + await flushPromises(); + + const healedWs = latestWs(); + expect(healedWs).not.toBe(firstWs); + healedWs.open(); + await flushPromises(); + await sendPromise; + + expect(resubscribeSpy).toHaveBeenCalled(); + expect(subscribeMessages(healedWs).length).toBeGreaterThanOrEqual(1); + expect( + subscribeMessages(healedWs).some((msg: any) => msg.streams?.[0]?.name === ScryptMessageType.BALANCE_TRANSACTION), + ).toBe(true); + + // TRADE notify was also sent on the healed socket. + const tradeSends = healedWs.send.mock.calls + .map(([payload]) => JSON.parse(payload as string)) + .filter((msg) => msg.type === ScryptMessageType.TRADE); + expect(tradeSends).toHaveLength(1); + }); + + it('does not schedule reconnect after an intentional disconnect', async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(); + const constructCountAfterConnect = WebSocket.instances.length; + + await connection.disconnect(); + + // Real ws emits 'close' after .close() completes — disconnect already set DISCONNECTED, + // so wasConnected is false and the reconnect block must not run. + firstWs.remoteClose(1000, 'normal closure'); + await flushPromises(); + + expect(scheduleSpy).not.toHaveBeenCalled(); + expect((connection as any).isReconnecting).toBe(false); + + jest.advanceTimersByTime(60000 * 3); + await flushPromises(); + + expect(WebSocket.instances.length).toBe(constructCountAfterConnect); + expect(scheduleSpy).not.toHaveBeenCalled(); + }); + + it('cancels a pending reconnect timer on intentional disconnect', async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(); + const constructCountAfterConnect = WebSocket.instances.length; + + // Unexpected drop schedules a reconnect timer; disconnect must cancel it before it fires. + firstWs.remoteClose(1006, 'gone'); + expect(scheduleSpy).toHaveBeenCalledWith(0, expect.any(Number)); + expect((connection as any).isReconnecting).toBe(true); + expect((connection as any).reconnectTimer).toBeDefined(); + + await connection.disconnect(); + + expect((connection as any).isReconnecting).toBe(false); + expect((connection as any).reconnectTimer).toBeUndefined(); + + jest.advanceTimersByTime(60000 * 3); + await flushPromises(); + + expect(WebSocket.instances.length).toBe(constructCountAfterConnect); + // No further scheduleReconnect from a fired timer (only the original attempt-0 schedule). + expect(scheduleSpy).toHaveBeenCalledTimes(1); + }); + + it('shared readiness: joining caller waits for resubscription before send (finding 1)', async () => { + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + firstWs.remoteClose(1006, 'gone'); + + let releaseResubscribe!: () => void; + const resubscribeGate = new Promise((resolve) => { + releaseResubscribe = resolve; + }); + const originalResubscribe = (connection as any).resubscribeToStreams.bind(connection); + jest.spyOn(connection as any, 'resubscribeToStreams').mockImplementation(async () => { + await resubscribeGate; + return originalResubscribe(); + }); + + await fireReconnectAttempt(0); + const reconnectedWs = latestWs(); + reconnectedWs.open(); + await flushPromises(); + + // Socket is open but resubscription is gated — still CONNECTING, not fully ready. + expect((connection as any).connectionState).toBe('connecting'); + expect(subscribeMessages(reconnectedWs)).toHaveLength(0); + + // Business caller joins the in-flight connect via ensureConnected → connectionPromise. + let sendResolved = false; + const sendPromise = connection.send(ScryptMessageType.TRADE, [{ side: 'buy' }]).then(() => { + sendResolved = true; + }); + await flushPromises(); + + expect(sendResolved).toBe(false); + expect((connection as any).connectionState).toBe('connecting'); + // TRADE must not be sent until streams are restored. + const tradeSendsBefore = reconnectedWs.send.mock.calls + .map(([payload]) => JSON.parse(payload as string)) + .filter((msg) => msg.type === ScryptMessageType.TRADE); + expect(tradeSendsBefore).toHaveLength(0); + + releaseResubscribe(); + await flushPromises(); + await sendPromise; + + expect((connection as any).connectionState).toBe('connected'); + expect(sendResolved).toBe(true); + + const allSends = reconnectedWs.send.mock.calls.map(([payload]) => JSON.parse(payload as string)); + const firstSubscribeIdx = allSends.findIndex((msg) => msg.type === 'subscribe'); + const tradeIdx = allSends.findIndex((msg) => msg.type === ScryptMessageType.TRADE); + expect(firstSubscribeIdx).toBeGreaterThanOrEqual(0); + expect(tradeIdx).toBeGreaterThan(firstSubscribeIdx); + expect(allSends[firstSubscribeIdx]).toEqual( + expect.objectContaining({ + type: 'subscribe', + streams: [expect.objectContaining({ name: ScryptMessageType.BALANCE_TRANSACTION })], + }), + ); + }); + + it('rejects connect and retries when the socket drops mid-resubscribe (finding 2)', async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + + firstWs.remoteClose(1006, 'gone'); + expect(scheduleSpy).toHaveBeenCalledWith(0, expect.any(Number)); + expect((connection as any).isReconnecting).toBe(true); + + await fireReconnectAttempt(0); + const reconnectedWs = latestWs(); + + // Emit a real close during resubscribe so handleDisconnection actually runs (not just a + // readyState flip that only trips assertSocketOpen). Mid-establish → CONNECTING must stay. + let stateAtDrop: string | undefined; + reconnectedWs.send.mockImplementation(() => { + reconnectedWs.remoteClose(1006, 'drop mid-resubscribe'); + // handleDisconnection is sync on emit; Fix A leaves CONNECTING (was not CONNECTED). + // Captured here (not asserted) because this callback runs inside resubscribeToStreams' + // try/catch, which would otherwise swallow a failing expect() and make it vacuous. + stateAtDrop = (connection as any).connectionState; + }); + + reconnectedWs.open(); + await flushPromises(); + + // connect() must reject → no success log; isReconnecting stays true; next attempt scheduled. + // Final DISCONNECTED comes from connect()'s catch after establishConnection rejects, not from + // handleDisconnection (which left CONNECTING — asserted above via stateAtDrop). + expect(stateAtDrop).toBe('connecting'); + expect(loggerInfo).not.toHaveBeenCalledWith(expect.stringMatching(/reconnected/)); + expect((connection as any).isReconnecting).toBe(true); + expect((connection as any).connectionState).toBe('disconnected'); + expect(loggerWarn).toHaveBeenCalledWith(expect.stringMatching(/reconnect attempt 1 failed/), expect.any(Error)); + expect(scheduleSpy).toHaveBeenCalledWith(1, expect.any(Number)); + + // Active stream must be kept for the next reconnect retry. + expect((connection as any).activeStreams.has(ScryptMessageType.BALANCE_TRANSACTION)).toBe(true); + + // Next backoff attempt is scheduled and can succeed. + await fireReconnectAttempt(1); + const retryWs = latestWs(); + expect(retryWs).not.toBe(reconnectedWs); + retryWs.open(); + await flushPromises(); + + expect(loggerInfo).toHaveBeenCalledWith(expect.stringMatching(/reconnected \(after 2 attempt/)); + expect((connection as any).isReconnecting).toBe(false); + expect( + subscribeMessages(retryWs).some((msg: any) => msg.streams?.[0]?.name === ScryptMessageType.BALANCE_TRANSACTION), + ).toBe(true); + }); + + it('clears stale reconnect state when a business call heals the socket (finding 1 — heal clears isReconnecting)', async () => { + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + + firstWs.remoteClose(1006, 'gone'); + expect((connection as any).isReconnecting).toBe(true); + expect((connection as any).reconnectTimer).toBeDefined(); + + // Before the backoff timer fires, a business call heals via ensureConnected. + const sendPromise = connection.send(ScryptMessageType.TRADE, [{ side: 'buy' }]); + await flushPromises(); + latestWs().open(); + await flushPromises(); + await sendPromise; + + expect((connection as any).connectionState).toBe('connected'); + expect((connection as any).isReconnecting).toBe(false); + expect((connection as any).reconnectTimer).toBeUndefined(); + + // A second drop on the healed socket must start a fresh reconnect loop (not skipped by stale true). + loggerWarn.mockClear(); + latestWs().remoteClose(1006, 'gone again'); + expect((connection as any).isReconnecting).toBe(true); + expect(loggerWarn).toHaveBeenCalledWith(expect.stringMatching(/closed.*scheduling reconnect/)); + }); + + it('disconnect() resets hasEverConnected so a later re-subscribe sends only one SUBSCRIBE (finding 2 — no double-send on reuse)', async () => { + const streamName = ScryptMessageType.BALANCE_TRANSACTION; + await firstConnectWithStream(streamName); + + await connection.disconnect(); + + connection.subscribeToStream(streamName, () => undefined); + const newWs = latestWs(); + newWs.open(); + await flushPromises(); + + const streamSubs = subscribeMessages(newWs).filter((msg: any) => + msg.streams?.some((s: any) => s.name === streamName), + ); + expect(streamSubs).toHaveLength(1); + expect(streamSubs[0]).toEqual( + expect.objectContaining({ + type: 'subscribe', + streams: [expect.objectContaining({ name: streamName })], + }), + ); + }); + + it('mid-establish drop does not spawn a second establishConnection (concurrency)', async () => { + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + firstWs.remoteClose(1006, 'gone'); + + // Gate resubscribe so we can observe the CONNECTING window after a real mid-establish close. + let releaseResubscribe!: () => void; + const resubscribeGate = new Promise((resolve) => { + releaseResubscribe = resolve; + }); + const originalResubscribe = (connection as any).resubscribeToStreams.bind(connection); + jest.spyOn(connection as any, 'resubscribeToStreams').mockImplementation(async () => { + await resubscribeGate; + return originalResubscribe(); + }); + + await fireReconnectAttempt(0); + const reconnectedWs = latestWs(); + reconnectedWs.open(); + await flushPromises(); + + // Handshake done, resubscribe gated — still CONNECTING with an in-flight connectionPromise. + expect((connection as any).connectionState).toBe('connecting'); + expect((connection as any).connectionPromise).toBeDefined(); + const inFlightPromise = (connection as any).connectionPromise; + const constructCountBeforeClose = WebSocket.instances.length; + + // Real close mid-establish: Fix A leaves CONNECTING (was not CONNECTED). + reconnectedWs.remoteClose(1006, 'drop mid-establish'); + expect((connection as any).connectionState).toBe('connecting'); + expect((connection as any).connectionPromise).toBe(inFlightPromise); + expect((connection as any).ws).toBeUndefined(); + + // Concurrent business call must join the same in-flight promise — no second WebSocket. + const sendPromise = connection.send(ScryptMessageType.TRADE, [{ side: 'buy' }]); + await flushPromises(); + + expect(WebSocket.instances.length).toBe(constructCountBeforeClose); + expect((connection as any).connectionState).toBe('connecting'); + expect((connection as any).connectionPromise).toBe(inFlightPromise); + + // Let establishConnection finish: resubscribe sees dead socket, assertSocketOpen rejects. + releaseResubscribe(); + await flushPromises(); + + await expect(sendPromise).rejects.toThrow(); + expect((connection as any).connectionState).toBe('disconnected'); + expect((connection as any).connectionPromise).toBeUndefined(); + // Still no extra socket constructed during the concurrent join window. + expect(WebSocket.instances.length).toBe(constructCountBeforeClose); + }); + + it('resubscribes all active streams on reconnect (multi-stream)', async () => { + const streams = [ScryptMessageType.BALANCE_TRANSACTION, ScryptMessageType.BALANCE, ScryptMessageType.TRADE]; + const firstWs = await firstConnectWithStreams(streams); + expect(subscribeMessages(firstWs)).toHaveLength(streams.length); + + firstWs.remoteClose(1006, 'gone'); + await fireReconnectAttempt(0); + const reconnectedWs = latestWs(); + reconnectedWs.open(); + await flushPromises(); + + const resubNames = subscribeMessages(reconnectedWs).map((msg: any) => msg.streams?.[0]?.name); + expect(resubNames).toHaveLength(streams.length); + for (const stream of streams) { + expect(resubNames).toContain(stream); + } + expect((connection as any).isReconnecting).toBe(false); + }); + + it('caps reconnect backoff at 60s with equal jitter in [capped/2, capped]', async () => { + const setTimeoutSpy = jest.spyOn(global, 'setTimeout'); + + // attempt >= 4 → min(5000 * 2**4, 60000) = 60000 + (connection as any).scheduleReconnect(4, (connection as any).reconnectEpoch); + + const reconnectTimerCalls = setTimeoutSpy.mock.calls.filter( + (call) => typeof call[1] === 'number' && (call[1] as number) >= 1000, + ); + expect(reconnectTimerCalls.length).toBeGreaterThanOrEqual(1); + const delay = reconnectTimerCalls[reconnectTimerCalls.length - 1][1] as number; + + expect(delay).toBeGreaterThanOrEqual(30000); + expect(delay).toBeLessThanOrEqual(60000); + }); + + it('rejects connect on handshake timeout and schedules the next backoff attempt', async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(); + + firstWs.remoteClose(1006, 'gone'); + expect(scheduleSpy).toHaveBeenCalledWith(0, expect.any(Number)); + + await fireReconnectAttempt(0); + const hungWs = latestWs(); + // Do not call open() — simulate a silent black-hole handshake. + expect(hungWs.readyState).toBe(WebSocket.CONNECTING); + + jest.advanceTimersByTime(15000); + await flushPromises(); + + expect(hungWs.terminate).toHaveBeenCalled(); + expect(loggerWarn).toHaveBeenCalledWith( + expect.stringMatching(/reconnect attempt 1 failed/), + expect.objectContaining({ message: expect.stringMatching(/handshake timed out after 15000ms/) }), + ); + expect(scheduleSpy).toHaveBeenCalledWith(1, expect.any(Number)); + expect((connection as any).isReconnecting).toBe(true); + expect(loggerInfo).not.toHaveBeenCalledWith(expect.stringMatching(/reconnected/)); + }); + + it("ignores a stale (superseded) socket's close (B3)", async () => { + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(); + + expect((connection as any).connectionGeneration).toBe(1); + expect((connection as any).ws).toBe(firstWs); + expect((connection as any).connectionState).toBe('connected'); + + // Simulate a close event from a prior attempt whose captured generation is no longer current. + (connection as any).handleDisconnection(0, 1006, 'stale'); + + expect((connection as any).ws).toBe(firstWs); + expect((connection as any).connectionState).toBe('connected'); + expect((connection as any).isReconnecting).toBe(false); + expect(scheduleSpy).not.toHaveBeenCalled(); + expect(loggerWarn).not.toHaveBeenCalledWith(expect.stringMatching(/scheduling reconnect/)); + }); + + it('disconnect() during pre-open handshake supersedes it — socket is terminated and not adopted (B1)', async () => { + const connectPromise = (connection as any).connect(); + await flushPromises(); + + expect((connection as any).connectionState).toBe('connecting'); + const preOpenWs = latestWs(); + expect(preOpenWs.readyState).toBe(WebSocket.CONNECTING); + + await connection.disconnect(); + + preOpenWs.open(); + await flushPromises(); + + expect(preOpenWs.terminate).toHaveBeenCalled(); + expect((connection as any).ws).toBeUndefined(); + expect((connection as any).connectionState).toBe('disconnected'); + await expect(connectPromise).rejects.toThrow(/superseded/i); + }); + + it('disconnect() during pre-open handshake then connect() starts a fresh attempt, not joining the superseded promise', async () => { + const firstConnectPromise = (connection as any).connect(); + await flushPromises(); + + expect((connection as any).connectionState).toBe('connecting'); + const preOpenWs = latestWs(); + expect(preOpenWs.readyState).toBe(WebSocket.CONNECTING); + const constructCountBeforeDisconnect = WebSocket.instances.length; + const generationBeforeDisconnect = (connection as any).connectionGeneration; + + await connection.disconnect(); + + // Fix: disconnect() resets state synchronously even though ws never opened. + expect((connection as any).connectionState).toBe('disconnected'); + expect((connection as any).connectionPromise).toBeUndefined(); + + // A fresh connect() call, issued before the pre-open socket ever settles, must start a NEW + // attempt — not join the doomed promise from the superseded attempt. + const secondConnectPromise = (connection as any).connect(); + await flushPromises(); + + expect(WebSocket.instances.length).toBe(constructCountBeforeDisconnect + 1); + expect((connection as any).connectionGeneration).toBeGreaterThan(generationBeforeDisconnect); + expect(secondConnectPromise).not.toBe(firstConnectPromise); + + const freshWs = latestWs(); + expect(freshWs).not.toBe(preOpenWs); + freshWs.open(); + await flushPromises(); + + await expect(secondConnectPromise).resolves.toBeUndefined(); + expect((connection as any).connectionState).toBe('connected'); + + // The stale pre-open socket, when it eventually opens, must still be rejected/terminated and + // must not clobber the fresh connection now in place. + preOpenWs.open(); + await flushPromises(); + await expect(firstConnectPromise).rejects.toThrow(/superseded/i); + expect(preOpenWs.terminate).toHaveBeenCalled(); + expect((connection as any).ws).toBe(freshWs); + expect((connection as any).connectionState).toBe('connected'); + }); + + it('disconnect() stops the backoff loop — no further reconnect after advancing timers (B2)', async () => { + const firstWs = await firstConnectWithStream(); + firstWs.remoteClose(1006, 'gone'); + expect((connection as any).isReconnecting).toBe(true); + + const constructCountBeforeDisconnect = WebSocket.instances.length; + await connection.disconnect(); + expect((connection as any).isReconnecting).toBe(false); + + // Re-arm a timer with a STALE epoch so the setTimeout callback's epoch guard is exercised + // (not merely clearTimeout of a still-pending timer). disconnect() already bumped reconnectEpoch. + const staleEpoch = (connection as any).reconnectEpoch - 1; + (connection as any).scheduleReconnect(0, staleEpoch); + jest.advanceTimersByTime(60000 * 3); + await flushPromises(); + + expect(WebSocket.instances.length).toBe(constructCountBeforeDisconnect); + expect((connection as any).isReconnecting).toBe(false); + }); + + it('stale reconnect loop settle does not disturb a newer loop after disconnect + drop', async () => { + // Prove Fix 3: a superseded reconnect loop's late-settling connect() must not clear + // isReconnecting or log "reconnected" for a newer live loop started after disconnect + drop. + const scheduleSpy = jest.spyOn(connection as any, 'scheduleReconnect'); + const firstWs = await firstConnectWithStream(ScryptMessageType.BALANCE_TRANSACTION); + + // Start loop A via unexpected drop. + firstWs.remoteClose(1006, 'gone'); + expect((connection as any).isReconnecting).toBe(true); + expect(scheduleSpy).toHaveBeenCalledWith(0, expect.any(Number)); + const epochA = scheduleSpy.mock.calls[0][1] as number; + + // Gate connect() so loop A's attempt stays in-flight. + let releaseConnectA!: () => void; + const connectAGate = new Promise((resolve) => { + releaseConnectA = resolve; + }); + let connectCallCount = 0; + const connectSpy = jest.spyOn(connection as any, 'connect').mockImplementation(async () => { + connectCallCount += 1; + if (connectCallCount === 1) { + await connectAGate; + return; + } + throw new Error('unexpected extra connect while loop A spy is active'); + }); + + await fireReconnectAttempt(0); + expect(connectCallCount).toBe(1); + expect((connection as any).isReconnecting).toBe(true); + + // disconnect() supersedes loop A: bumps epoch, clears isReconnecting, cancels timer. + await connection.disconnect(); + expect((connection as any).isReconnecting).toBe(false); + expect((connection as any).reconnectEpoch).toBeGreaterThan(epochA); + + // Fresh connection + drop starts loop B with a new epoch. + connectSpy.mockRestore(); + connection.subscribeToStream(ScryptMessageType.BALANCE_TRANSACTION, () => undefined); + await flushPromises(); + const secondWs = latestWs(); + secondWs.open(); + await flushPromises(); + expect((connection as any).connectionState).toBe('connected'); + + const scheduleCountBeforeLoopB = scheduleSpy.mock.calls.length; + secondWs.remoteClose(1006, 'drop again'); + expect((connection as any).isReconnecting).toBe(true); + expect(scheduleSpy.mock.calls.length).toBe(scheduleCountBeforeLoopB + 1); + const epochB = scheduleSpy.mock.calls[scheduleSpy.mock.calls.length - 1][1] as number; + expect(epochB).toBeGreaterThan(epochA); + expect(epochB).toBe((connection as any).reconnectEpoch); + + const constructCountWithLoopBPending = WebSocket.instances.length; + const timerAfterLoopB = (connection as any).reconnectTimer; + expect(timerAfterLoopB).toBeDefined(); + + // Let loop A's gated connect finally resolve — stale .then must no-op. + loggerInfo.mockClear(); + releaseConnectA(); + await flushPromises(); + + expect((connection as any).isReconnecting).toBe(true); + expect((connection as any).reconnectEpoch).toBe(epochB); + expect((connection as any).reconnectTimer).toBe(timerAfterLoopB); + expect(loggerInfo).not.toHaveBeenCalledWith(expect.stringMatching(/reconnected/)); + expect(scheduleSpy.mock.calls.length).toBe(scheduleCountBeforeLoopB + 1); + expect(WebSocket.instances.length).toBe(constructCountWithLoopBPending); + }); +}); diff --git a/src/integration/exchange/services/scrypt-websocket-connection.ts b/src/integration/exchange/services/scrypt-websocket-connection.ts index bf273bd07b..f981995315 100644 --- a/src/integration/exchange/services/scrypt-websocket-connection.ts +++ b/src/integration/exchange/services/scrypt-websocket-connection.ts @@ -71,8 +71,15 @@ export class ScryptWebSocketConnection { private ws?: WebSocket; private connectionState: ConnectionState = ConnectionState.DISCONNECTED; private connectionPromise?: Promise; + private connectionGeneration = 0; private readonly reconnectDelay = 5000; // 5 seconds + private readonly maxReconnectDelay = 60000; // 60s cap for the exponential backoff + private readonly handshakeTimeoutMs = 15000; + private hasEverConnected = false; // first connect subscribes directly; later connects must resubscribe + private isReconnecting = false; // guards against overlapping reconnect loops + private reconnectEpoch = 0; // bumped on disconnect / new loop so stale scheduleReconnect continuations no-op + private reconnectTimer?: NodeJS.Timeout; // requests private reqIdCounter = 0; @@ -183,44 +190,91 @@ export class ScryptWebSocketConnection { } async disconnect(): Promise { - if (!this.ws) return; - + this.connectionGeneration++; // supersede any in-flight connect attempt (its socket events become no-ops) + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; + } + this.isReconnecting = false; + this.hasEverConnected = false; // full reset: a later reuse re-subscribes as a first connect (no double-send) + this.reconnectEpoch++; // supersede any in-flight reconnect loop (stale timer/then/catch become no-ops) this.connectionState = ConnectionState.DISCONNECTED; - this.ws.close(); - this.ws = undefined; + this.connectionPromise = undefined; this.pendingRequests.forEach((request) => { clearTimeout(request.timeout); request.reject(new Error('Connection closed')); }); this.pendingRequests.clear(); - this.subscriptions.clear(); this.activeStreams.clear(); + + if (this.ws) { + this.ws.close(); + this.ws = undefined; + } } // --- CONNECTION MANAGEMENT --- // private async connect(): Promise { if (this.connectionState === ConnectionState.CONNECTED) return; + if (this.connectionState === ConnectionState.CONNECTING && this.connectionPromise) return this.connectionPromise; - if (this.connectionState === ConnectionState.CONNECTING && this.connectionPromise) { - return this.connectionPromise; - } - + const generation = ++this.connectionGeneration; this.connectionState = ConnectionState.CONNECTING; - this.connectionPromise = this.connectWebSocket(); - + const promise = this.establishConnection(generation); + this.connectionPromise = promise; try { - await this.connectionPromise; - this.connectionState = ConnectionState.CONNECTED; + await promise; } catch (error) { - this.connectionState = ConnectionState.DISCONNECTED; - this.connectionPromise = undefined; + // Only the owner of the current in-flight promise may reset state — a late reject from a + // stale establish must not clobber a newer attempt that already installed its own promise. + if (this.connectionPromise === promise) { + this.connectionState = ConnectionState.DISCONNECTED; + this.connectionPromise = undefined; + } throw error; } } + // Full readiness. Everything awaiting connect()/connectionPromise waits for ALL of this, so no caller can send + // on a socket whose streams are not yet restored (#4310 finding 1); a drop at any step rejects so the caller / + // backoff loop retries instead of treating a dead socket as connected (finding 2). connectionState stays + // CONNECTING until the very end, so a business call arriving mid-resubscribe joins connectionPromise (via + // connect()'s CONNECTING branch) rather than proceeding on the half-ready socket. + private async establishConnection(generation: number): Promise { + // rejects on error or handshake timeout; a close-before-open without an error is bounded by the + // handshake timeout (it does not itself reject) + await this.connectWebSocket(generation); + this.assertCurrentGeneration(generation); + this.assertSocketOpen('after handshake'); + + if (this.hasEverConnected) { + await this.resubscribeToStreams(); // sends on the current socket directly, no ensureConnected (no reentrancy) + this.assertCurrentGeneration(generation); + this.assertSocketOpen('after resubscription'); + } else { + this.hasEverConnected = true; + } + + this.connectionState = ConnectionState.CONNECTED; // fully ready only now + // We are fully connected — clear any reconnect loop, whether it healed us or a business call did. + this.isReconnecting = false; + if (this.reconnectTimer) { + clearTimeout(this.reconnectTimer); + this.reconnectTimer = undefined; + } + } + + private assertCurrentGeneration(generation: number): void { + if (generation !== this.connectionGeneration) throw new Error('Scrypt WebSocket connection attempt superseded'); + } + + private assertSocketOpen(when: string): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) throw new Error(`Scrypt WebSocket is not open ${when}`); + } + private async ensureConnected(): Promise { if (this.ws && this.ws.readyState === WebSocket.OPEN && this.connectionState === ConnectionState.CONNECTED) { return this.ws; @@ -235,7 +289,7 @@ export class ScryptWebSocketConnection { return this.ws; } - private async connectWebSocket(): Promise { + private async connectWebSocket(generation: number): Promise { return new Promise((resolve, reject) => { const url = new URL(this.wsUrl); const host = url.host; @@ -254,8 +308,18 @@ export class ScryptWebSocketConnection { }; const ws = new WebSocket(this.wsUrl, { headers }); + const handshakeTimeout = setTimeout(() => { + ws.terminate(); + reject(new Error(`Scrypt WebSocket handshake timed out after ${this.handshakeTimeoutMs}ms`)); + }, this.handshakeTimeoutMs); ws.on('open', () => { + clearTimeout(handshakeTimeout); + if (generation !== this.connectionGeneration) { + ws.terminate(); // a newer attempt or disconnect() superseded us — do not adopt this socket + reject(new Error('Scrypt WebSocket connection attempt superseded')); + return; + } this.ws = ws; resolve(); }); @@ -265,19 +329,25 @@ export class ScryptWebSocketConnection { }); ws.on('error', (error) => { + clearTimeout(handshakeTimeout); this.logger.error('Scrypt WebSocket error:', error); reject(error); }); ws.on('close', (code, reason) => { - this.handleDisconnection(code, reason); + this.handleDisconnection(generation, code, reason); }); }); } - private handleDisconnection(code?: number, reason?: string): void { + private handleDisconnection(generation: number, code?: number, reason?: string): void { + if (generation !== this.connectionGeneration) return; // stale socket — its close is not our concern + + // Only flip to DISCONNECTED when the live socket dropped. A mid-establish close (CONNECTING) + // must leave state alone so concurrent connect() callers still join connectionPromise instead + // of starting a second establishConnection. const wasConnected = this.connectionState === ConnectionState.CONNECTED; - this.connectionState = ConnectionState.DISCONNECTED; + if (wasConnected) this.connectionState = ConnectionState.DISCONNECTED; this.ws = undefined; // reject pending requests @@ -288,19 +358,35 @@ export class ScryptWebSocketConnection { this.pendingRequests.clear(); // reconnect - if (wasConnected) { - this.logger.warn( - `Scrypt WebSocket closed (code: ${code}, reason: ${reason}), attempting reconnect in ${this.reconnectDelay}ms`, - ); - - setTimeout(() => { - void this.connect() - .then(() => this.resubscribeToStreams()) - .catch((error) => this.logger.error('Reconnection failed:', error)); - }, this.reconnectDelay); + if (wasConnected && !this.isReconnecting) { + this.isReconnecting = true; + const epoch = ++this.reconnectEpoch; + this.logger.warn(`Scrypt WebSocket closed (code: ${code}, reason: ${reason}), scheduling reconnect`); + this.scheduleReconnect(0, epoch); } } + private scheduleReconnect(attempt: number, epoch: number): void { + const capped = Math.min(this.reconnectDelay * 2 ** attempt, this.maxReconnectDelay); + const delay = capped / 2 + Math.random() * (capped / 2); // equal jitter to avoid synchronized retry storms + if (attempt > 0 && attempt % 10 === 0) + this.logger.error(`Scrypt WebSocket still not reconnected after ${attempt} attempts`); + this.reconnectTimer = setTimeout(() => { + if (epoch !== this.reconnectEpoch) return; // this loop was superseded (disconnect or a newer loop) + void this.connect() + .then(() => { + if (epoch !== this.reconnectEpoch) return; + this.isReconnecting = false; + this.logger.info(`Scrypt WebSocket reconnected (after ${attempt + 1} attempt(s))`); + }) + .catch((error) => { + if (epoch !== this.reconnectEpoch) return; + this.logger.warn(`Scrypt WebSocket reconnect attempt ${attempt + 1} failed; retrying`, error); + this.scheduleReconnect(attempt + 1, epoch); + }); + }, delay); + } + // --- REQUEST/RESPONSE --- // private async notify(message: ScryptRequest): Promise { @@ -352,6 +438,14 @@ export class ScryptWebSocketConnection { // --- STREAMING SUBSCRIPTIONS --- // + /** + * Safe to call for a stream that is already active (it only adds a callback — no new SUBSCRIBE is + * sent). Subscribing a brand-new stream while the connection is down/reconnecting can double-send + * the SUBSCRIBE frame (the in-flight reconnect's resubscribe and this call both send it). All + * current callers either subscribe at construction or, like + * ExchangeTxService.onBalanceTransactions(), only add a callback to an already-active stream — see + * #4310 follow-up. + */ subscribeToStream( streamName: ScryptMessageType, callback: (data: T[]) => void, @@ -400,20 +494,22 @@ export class ScryptWebSocketConnection { } private async sendSubscription(streamName: ScryptMessageType, filters?: Record): Promise { - const ws = await this.ensureConnected(); - - const request: ScryptRequest = { - reqid: ++this.reqIdCounter, - type: ScryptRequestType.SUBSCRIBE, - streams: [ - { - name: streamName, - ...filters, - }, - ], - }; + await this.ensureConnected(); + this.sendSubscriptionOnSocket(streamName, filters); + } - ws.send(JSON.stringify(request)); + // Send a SUBSCRIBE frame on the CURRENT socket, used during (re)connection where ensureConnected must not be + // called (connectionState is still CONNECTING). Throws if the socket is not open so a mid-resubscribe drop is + // detectable by establishConnection's assertSocketOpen. + private sendSubscriptionOnSocket(streamName: ScryptMessageType, filters?: Record): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) throw new Error('Scrypt WebSocket is not open'); + this.ws.send( + JSON.stringify({ + reqid: ++this.reqIdCounter, + type: ScryptRequestType.SUBSCRIBE, + streams: [{ name: streamName, ...filters }], + }), + ); } private async sendUnsubscription(streamName: ScryptMessageType): Promise { @@ -444,15 +540,12 @@ export class ScryptWebSocketConnection { } private async resubscribeToStreams(): Promise { - const streams = Array.from(this.activeStreams); - this.activeStreams.clear(); - - for (const streamName of streams) { + for (const streamName of this.activeStreams) { try { - await this.sendSubscription(streamName); - this.activeStreams.add(streamName); + this.sendSubscriptionOnSocket(streamName); // throws if the socket isn't open; caught + logged, kept for retry } catch (error) { this.logger.error(`Failed to resubscribe to ${streamName}:`, error); + // keep it in activeStreams so the next reconnect retries it (do not drop it) } } } From 56a197cf999cbb857aa1c592497621ba75117afc Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:52:19 +0200 Subject: [PATCH 5/5] chore(liquidity): reduce lmActivationDelay from 15 to 10 minutes (#4333) Prod-only data migration lowering the global lmActivationDelay runtime setting from 15 to 10. This is the debounce window a liquidity-management rule's deficit/redundancy condition must persist before a fund-moving pipeline starts; it is global (affects all liquidity rules). Guarded to ENVIRONMENT === 'prd' (the setting row may be absent on dev/loc/CI). up() is fail-loud on row existence and the '10' post-state; the prior value is intentionally not asserted because the setting is runtime-mutable. down() best-effort restores 15. --- .../1784799156042-ReduceLmActivationDelay.js | 105 ++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 migration/1784799156042-ReduceLmActivationDelay.js diff --git a/migration/1784799156042-ReduceLmActivationDelay.js b/migration/1784799156042-ReduceLmActivationDelay.js new file mode 100644 index 0000000000..b2e93bdf8c --- /dev/null +++ b/migration/1784799156042-ReduceLmActivationDelay.js @@ -0,0 +1,105 @@ +/** + * @typedef {import('typeorm').MigrationInterface} MigrationInterface + * @typedef {import('typeorm').QueryRunner} QueryRunner + */ + +/** + * PROD-ONLY data migration that reduces the global `lmActivationDelay` runtime setting from 15 to 10 + * minutes in production. + * + * Guarded to `ENVIRONMENT === 'prd'` for the same reason as the sibling liquidity data migrations + * (e.g. ReactivateEthZchfLiquidityAndEnableMexcZchfSell): liquidity-management behaviour must not be + * mutated on dev/loc/CI. Additionally, the `setting` row for `lmActivationDelay` may not exist at + * all on those environments (fresh/seeded database) - the row-exists precondition would otherwise + * crash boot there. On prod the row exists. Returning early still records the migration as executed, + * which is the intended no-op on lower environments. + * + * What the setting is: + * `lmActivationDelay` is the debounce window (in minutes) a LiquidityManagementRule's + * deficit/redundancy condition must persist before a fund-moving pipeline starts. It is read in + * `src/subdomains/core/liquidity-management/services/liquidity-management.service.ts` via + * `this.settingService.get('lmActivationDelay', '30')`. It is global - it affects ALL liquidity + * rules, not a single asset. + * + * Prior-value policy: + * The exact prior value is deliberately NOT asserted (e.g. no hard requirement that value === '15'). + * `lmActivationDelay` is a runtime-mutable setting; a strict prior-value assertion would turn a + * benign runtime change into a boot-crash on the next deploy. Only the row-exists precondition and + * the target post-condition (value === '10') are enforced fail-loud. + * + * The lock_timeout is set transaction-scoped via `SET LOCAL lock_timeout` (consistent with the Frick + * and sibling LM reference migrations); TypeORM runs up()/down() each inside its own transaction. + * + * up(): + * 1. prod guard (no-op elsewhere) + * 2. lock_timeout + * 3. fail-loud precondition: row key='lmActivationDelay' exists (value not asserted) + * 4. UPDATE setting value -> '10' + * 5. fail-loud post-condition: value is exactly '10' + * + * down() best-effort restores value='15' (the value at authoring time). This is NOT a guaranteed + * inverse, because the setting is runtime-mutable and may have been changed after up() ran. No + * preconditions in down(). Prod-guarded the same way. + * + * @class + * @implements {MigrationInterface} + */ +module.exports = class ReduceLmActivationDelay1784799156042 { + name = 'ReduceLmActivationDelay1784799156042'; + + /** + * @param {QueryRunner} queryRunner + */ + async up(queryRunner) { + // Liquidity-management settings must NEVER be mutated on dev/loc/CI, and the lmActivationDelay + // row may not exist there at all (fresh/seeded DB) - the row-exists precondition would crash boot. + // Returning early still records the migration as executed, which is the intended no-op on lower + // environments. + if (process.env.ENVIRONMENT !== 'prd') return; + + await queryRunner.query(`SET LOCAL lock_timeout = '5s'`); + + // --- Precondition: row exists (value deliberately not asserted; setting is runtime-mutable) --- + const before = ( + await queryRunner.query(` + SELECT "value" FROM "setting" WHERE "key" = 'lmActivationDelay' + `) + ).at(0); + if (!before) { + throw new Error( + "Precondition failed: setting row with key='lmActivationDelay' not found", + ); + } + + await queryRunner.query(` + UPDATE "setting" SET "value" = '10' WHERE "key" = 'lmActivationDelay' + `); + + // --- Post-condition: target value must be exactly '10' --- + const after = ( + await queryRunner.query(` + SELECT "value" FROM "setting" WHERE "key" = 'lmActivationDelay' + `) + ).at(0); + if (!after || after.value !== '10') { + throw new Error( + `Post-condition failed for lmActivationDelay: expected value='10'; got value=${after && after.value}`, + ); + } + } + + /** + * @param {QueryRunner} queryRunner + */ + async down(queryRunner) { + if (process.env.ENVIRONMENT !== 'prd') return; + + await queryRunner.query(`SET LOCAL lock_timeout = '5s'`); + + // Best-effort restore to the value at authoring time ('15'). Not a guaranteed inverse: the + // setting is runtime-mutable and may have been changed after up() ran. No preconditions. + await queryRunner.query(` + UPDATE "setting" SET "value" = '15' WHERE "key" = 'lmActivationDelay' + `); + } +};