From 9c1f40b5fbfcc48c6f4504684761f49b4ba4e8b4 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:40:58 +0200 Subject: [PATCH 1/2] fix(scrypt): catch up state on reconnect and cancel leaked fetch streams (#4310) (#4335) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(scrypt): catch up state on reconnect and cancel leaked fetch streams (#4310) Completes the receive-path resilience epic: - Leak-fix: fetch/fetchAll opened a server-side stream via `type: subscribe` and never stopped it. They now cancel it by request id (the venue's cancel-by-reqid) after collecting the response — best-effort, never affecting the fetch result or triggering a retry. - Catch-up: a bare re-subscribe is not documented to replay a BalanceTransaction missed during an outage, so after a reconnect the connection fires an onReconnect hook and ScryptService re-fetches balance transactions + execution reports (as the constructor warm-up does), recovering the missed event within the reconnect instead of never. The balance-transaction cache write is terminal-aware, so a fetched non-terminal record can never clobber a fresher completed one. Stacked on the WS reconnect PR; the onReconnect callbacks are fire-and-forget (never awaited in establishConnection) so the catch-up cannot block or delay a reconnect. * fix(scrypt): apply the terminal cache guard universally, cancel on every fetch exit, coalesce catch-ups (#4310) Address review findings on the reconnect catch-up + leak-fix: - The permanent live BalanceTransaction subscriber and the constructor warm-up now write through cacheBalanceTransaction too, so the terminal guard is universal — a stale non-terminal push (e.g. a resubscribe snapshot on reconnect) can no longer clobber a completed withdrawal in the cache that getWithdrawalStatus reads. - fetch/fetchAll cancel their ad-hoc stream in a finally, so a malformed-initial throw, a request timeout, or a mid-pagination failure still stops the server-side stream instead of leaking it on the non-happy path. - catchUpAfterReconnect coalesces via an in-progress flag, so a re-drop during its own fetch cannot spawn a second concurrent catch-up (which could regress an execution report, as those carry no terminal guard). - fireReconnectCallbacks now also swallows async rejections from a callback, not just synchronous throws. * fix(scrypt): coalesce overlapping reconnect catch-ups instead of dropping them (#4310) The catchUpInProgress guard hard-skipped an overlapping reconnect, so on a double reconnect the already-resolved execution-report leg was never re-fetched and executionReports could be left stale for an order whose final transition landed in the second reconnect's downtime (getOrderStatus reads the cache with no staleness check). Replace the skip with a catchUpPending flag: an overlapping trigger makes the in-flight catch-up run once more (do/while), coalescing multiple overlaps into a single rerun that covers their downtime. * fix(scrypt): guard execution reports too, DRY the cache-apply, parallelize catch-up (#4310) pr-ready review: executionReports had no terminal guard while balanceTransactions did, and the reconnect catch-up now writes execution reports — so a live fill could be clobbered by a staler catch-up fetch, regressing an order (possible re-cancel/duplicate). Add isTerminalExecutionReport + cacheExecutionReport and route all three execution-report write sites (warm-up, live subscriber, catch-up) through it, mirroring balanceTransactions. Extract applyExecutionReports/applyBalanceTransactions helpers; run the catch-up's two independent fetches via Promise.allSettled; sendCancel logs at error severity. * test(scrypt): cover reconnect catch-up allSettled isolation and cancel-frame throw path (#4310) pr-ready round 2: add two coverage tests. (1) catchUpAfterReconnect applies a fulfilled stream even when the other stream's fetchAll rejects, and logs the rejection once (Promise.allSettled isolation). (2) fetch still resolves with its collected data when the best-effort CANCEL frame's ws.send throws synchronously, logging the failure. Test-only; no production change. * fix(scrypt): cache live stream updates directly so they are never age-filtered (#4310) pr-ready round 3: the DRY refactor routed the live EXECUTION_REPORT/BALANCE_TRANSACTION subscription callbacks through the same age-bounded helpers as the bulk warm-up/catch-up, so a live BalanceTransaction without a Timestamp (new Date(undefined) >= cutoff is false) was silently dropped. Live callbacks now cache each item directly via cacheExecutionReport/cacheBalanceTransaction (terminal guard, no age cutoff); applyExecutionReports/applyBalanceTransactions stay on the bulk paths only. Pinned by mutation-checked tests plus a bulk-path age-filter regression test. * fix(scrypt): route getOrderStatus fallback through the terminal guard and key-guard balance-tx cache (#4310) pr-ready round 4: getOrderStatus's cache-miss fallback wrote executionReports directly, bypassing the terminal-aware guard the other write paths use — a live terminal push landing during the fetch await could be clobbered by a stale non-terminal snapshot and, since a terminal order emits no further update, persist until restart. Route it through cacheExecutionReport and re-read the cache. Also guard cacheBalanceTransaction's terminal check on a defined ClReqID so it no longer suppresses distinct keyless records (ClReqID is optional). Both pinned by mutation-checked tests. --- .../scrypt-websocket-connection.spec.ts | 239 +++++++++ .../services/__tests__/scrypt.service.spec.ts | 459 ++++++++++++++++++ .../services/scrypt-websocket-connection.ts | 84 +++- .../exchange/services/scrypt.service.ts | 105 +++- 4 files changed, 846 insertions(+), 41 deletions(-) create mode 100644 src/integration/exchange/services/__tests__/scrypt.service.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 index 24c8d4de1f..1e35fa95ab 100644 --- a/src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts +++ b/src/integration/exchange/services/__tests__/scrypt-websocket-connection.spec.ts @@ -760,4 +760,243 @@ describe('ScryptWebSocketConnection', () => { expect(scheduleSpy.mock.calls.length).toBe(scheduleCountBeforeLoopB + 1); expect(WebSocket.instances.length).toBe(constructCountWithLoopBPending); }); + + it('fetchAll sends a cancel after collecting all pages', async () => { + const ws = await firstConnectWithStream(); + const streamName = ScryptMessageType.BALANCE_TRANSACTION; + + const fetchPromise = connection.fetchAll(streamName); + await flushPromises(); + + // First page (subscribe) — find the ad-hoc subscribe reqid (not the long-lived stream sub). + const page1Send = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .filter(({ msg }) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName) + .pop(); + expect(page1Send).toBeDefined(); + const reqId = page1Send!.msg.reqid as number; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + initial: true, + data: [{ id: 1 }], + next: 'cursor1', + }), + ); + await flushPromises(); + + // Second page (page request reuses the same reqid) + const page2Send = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .find(({ msg }) => msg.type === 'page' && msg.reqid === reqId); + expect(page2Send).toBeDefined(); + const page2Idx = page2Send!.idx; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + data: [{ id: 2 }], + }), + ); + await flushPromises(); + + const result = await fetchPromise; + expect(result).toEqual([{ id: 1 }, { id: 2 }]); + + const cancelSend = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .find(({ msg }) => msg.type === 'cancel' && msg.reqid === reqId); + expect(cancelSend).toBeDefined(); + expect(cancelSend!.msg).toEqual({ reqid: reqId, type: 'cancel' }); + // Cancel must be sent after the page-2 request (collection complete). + expect(cancelSend!.idx).toBeGreaterThan(page2Idx); + }); + + it('fetch sends a cancel after collecting its response', async () => { + const ws = await firstConnectWithStream(); + const streamName = ScryptMessageType.EXECUTION_REPORT; + + const fetchPromise = connection.fetch(streamName); + await flushPromises(); + + const subscribeSend = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .filter(({ msg }) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName) + .pop(); + expect(subscribeSend).toBeDefined(); + const reqId = subscribeSend!.msg.reqid as number; + const subscribeIdx = subscribeSend!.idx; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + initial: true, + data: [{ ClOrdID: 'ord-1' }], + }), + ); + await flushPromises(); + + const result = await fetchPromise; + expect(result).toEqual([{ ClOrdID: 'ord-1' }]); + + const cancelSend = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .find(({ msg }) => msg.type === 'cancel' && msg.reqid === reqId); + expect(cancelSend).toBeDefined(); + expect(cancelSend!.msg).toEqual({ reqid: reqId, type: 'cancel' }); + expect(cancelSend!.idx).toBeGreaterThan(subscribeIdx); + }); + + it('sendCancel is a no-op when the socket is not open', () => { + expect(WebSocket.instances.length).toBe(0); + expect((connection as any).ws).toBeUndefined(); + + expect(() => (connection as any).sendCancel(123)).not.toThrow(); + + expect(WebSocket.instances.length).toBe(0); + }); + + it('sendCancel is a no-op when the socket is present but not OPEN', async () => { + const ws = await firstConnectWithStream(); + const sendCallsBefore = ws.send.mock.calls.length; + + ws.readyState = WebSocket.CLOSING; + (connection as any).ws = ws; + + expect(() => (connection as any).sendCancel(456)).not.toThrow(); + + const cancelSends = ws.send.mock.calls + .slice(sendCallsBefore) + .map(([payload]) => JSON.parse(payload as string)) + .filter((msg) => msg.type === 'cancel'); + expect(cancelSends).toHaveLength(0); + }); + + it('fetch resolves with its collected data even when the cancel frame throws synchronously', async () => { + const ws = await firstConnectWithStream(); + const streamName = ScryptMessageType.EXECUTION_REPORT; + + const fetchPromise = connection.fetch(streamName); + await flushPromises(); + + const subscribeSend = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .filter(({ msg }) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName) + .pop(); + expect(subscribeSend).toBeDefined(); + const reqId = subscribeSend!.msg.reqid as number; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + initial: true, + data: [{ ClOrdID: 'ord-cancel-throw' }], + }), + ); + + // Scope the throw to the very next ws.send call — the CANCEL frame sent from fetch's finally block. + // Must be set synchronously right after emit, before any await drains the microtask queue. + ws.send.mockImplementationOnce(() => { + throw new Error('send failed'); + }); + + const result = await fetchPromise; + + expect(result).toEqual([{ ClOrdID: 'ord-cancel-throw' }]); + expect(loggerError).toHaveBeenCalledWith(`Failed to cancel Scrypt stream ${reqId}:`, expect.any(Error)); + }); + + it('fetch still sends cancel when the collect path throws (malformed initial)', async () => { + const ws = await firstConnectWithStream(); + const streamName = ScryptMessageType.EXECUTION_REPORT; + + const fetchPromise = connection.fetch(streamName); + await flushPromises(); + + const subscribeSend = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .filter(({ msg }) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName) + .pop(); + expect(subscribeSend).toBeDefined(); + const reqId = subscribeSend!.msg.reqid as number; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + initial: false, + data: [{ ClOrdID: 'ord-bad' }], + }), + ); + await flushPromises(); + + await expect(fetchPromise).rejects.toThrow(/Expected initial/); + + const cancelSend = ws.send.mock.calls + .map(([payload]) => JSON.parse(payload as string)) + .find((msg) => msg.type === 'cancel' && msg.reqid === reqId); + expect(cancelSend).toBeDefined(); + expect(cancelSend).toEqual({ reqid: reqId, type: 'cancel' }); + }); + + it('fetchAll still sends cancel when the collect path throws (malformed initial)', async () => { + const ws = await firstConnectWithStream(); + const streamName = ScryptMessageType.BALANCE_TRANSACTION; + + const fetchPromise = connection.fetchAll(streamName); + await flushPromises(); + + const page1Send = ws.send.mock.calls + .map(([payload], idx) => ({ msg: JSON.parse(payload as string), idx })) + .filter(({ msg }) => msg.type === 'subscribe' && msg.streams?.[0]?.name === streamName) + .pop(); + expect(page1Send).toBeDefined(); + const reqId = page1Send!.msg.reqid as number; + + ws.emit( + 'message', + JSON.stringify({ + reqid: reqId, + type: streamName, + // omit initial — malformed first page + data: [{ id: 1 }], + }), + ); + await flushPromises(); + + await expect(fetchPromise).rejects.toThrow(/Expected initial/); + + const cancelSend = ws.send.mock.calls + .map(([payload]) => JSON.parse(payload as string)) + .find((msg) => msg.type === 'cancel' && msg.reqid === reqId); + expect(cancelSend).toBeDefined(); + expect(cancelSend).toEqual({ reqid: reqId, type: 'cancel' }); + }); + + it('fires onReconnect callbacks on genuine reconnect but not on first connect', async () => { + const cb = jest.fn(); + connection.onReconnect(cb); + + await firstConnectWithStream(); + expect(cb).not.toHaveBeenCalled(); + + const firstWs = latestWs(); + firstWs.remoteClose(1006, 'gone'); + await fireReconnectAttempt(0); + const reconnectedWs = latestWs(); + reconnectedWs.open(); + await flushPromises(); + + expect(cb).toHaveBeenCalledTimes(1); + }); }); diff --git a/src/integration/exchange/services/__tests__/scrypt.service.spec.ts b/src/integration/exchange/services/__tests__/scrypt.service.spec.ts new file mode 100644 index 0000000000..a42e2d4ca6 --- /dev/null +++ b/src/integration/exchange/services/__tests__/scrypt.service.spec.ts @@ -0,0 +1,459 @@ +import { ScryptOrderStatus, ScryptTransactionStatus } from '../../dto/scrypt.dto'; +import { ScryptMessageType, ScryptWebSocketConnection } from '../scrypt-websocket-connection'; +import { ScryptService } from '../scrypt.service'; + +jest.mock('src/config/config', () => { + const mockConfig = { + scrypt: { + apiKey: 'k', + apiSecret: 's', + wsUrl: 'wss://x', + }, + }; + return { + Config: mockConfig, + GetConfig: () => mockConfig, + }; +}); + +jest.mock('../scrypt-websocket-connection', () => { + const actual = jest.requireActual('../scrypt-websocket-connection'); + return { + ...actual, + ScryptWebSocketConnection: jest.fn().mockImplementation(() => ({ + fetchAll: jest.fn().mockResolvedValue([]), + fetch: jest.fn().mockResolvedValue([]), + subscribeToStream: jest.fn().mockReturnValue(() => undefined), + onReconnect: jest.fn(), + send: jest.fn(), + requestAndWaitForUpdate: jest.fn(), + })), + }; +}); + +async function flushPromises(): Promise { + for (let i = 0; i < 30; i++) { + await Promise.resolve(); + } +} + +describe('ScryptService', () => { + let service: ScryptService; + let instance: { + fetchAll: jest.Mock; + onReconnect: jest.Mock; + subscribeToStream: jest.Mock; + }; + + beforeEach(async () => { + (ScryptWebSocketConnection as jest.MockedClass).mockClear(); + service = new ScryptService(); + instance = (ScryptWebSocketConnection as jest.MockedClass).mock.results[0] + .value as any; + // Constructor warm-up fetchAll calls settle on empty arrays before tests reconfigure. + await flushPromises(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + }); + + it('catchUpAfterReconnect fetches fresh state and applies terminal-aware balance-tx cache writes', async () => { + const now = new Date().toISOString(); + + const existingNonTerminal = { + ClReqID: 'a', + TransactionID: 'tx-a-old', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: now, + }; + const existingTerminal = { + ClReqID: 'b', + TransactionID: 'tx-b-terminal', + Status: ScryptTransactionStatus.REJECTED, + Timestamp: now, + }; + (service as any).balanceTransactions.set('a', existingNonTerminal); + (service as any).balanceTransactions.set('b', existingTerminal); + const terminalBBefore = (service as any).balanceTransactions.get('b'); + + const freshTerminalA = { + ClReqID: 'a', + TransactionID: 'tx-a-new', + Status: ScryptTransactionStatus.COMPLETED, + TxHash: 'hash1', + Timestamp: now, + }; + const freshNonTerminalB = { + ClReqID: 'b', + TransactionID: 'tx-b-nonterminal', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: now, + }; + const freshReport = { + ClOrdID: 'ord-1', + SubmitTime: now, + OrderID: 'oid-1', + }; + + instance.fetchAll.mockImplementation(async (streamName: string) => { + if (streamName === ScryptMessageType.BALANCE_TRANSACTION) { + return [freshTerminalA, freshNonTerminalB]; + } + if (streamName === ScryptMessageType.EXECUTION_REPORT) { + return [freshReport]; + } + return []; + }); + + await (service as any).catchUpAfterReconnect(); + + expect((service as any).balanceTransactions.get('a')).toEqual(freshTerminalA); + expect((service as any).balanceTransactions.get('b')).toBe(terminalBBefore); + expect((service as any).balanceTransactions.get('b')).toEqual(existingTerminal); + expect((service as any).executionReports.get('ord-1')).toEqual(freshReport); + }); + + it('cacheBalanceTransaction allows terminal→terminal correction (does not block all updates once terminal)', () => { + const now = new Date().toISOString(); + const existingTerminal = { + ClReqID: 'c', + TransactionID: 'tx-c-old', + Status: ScryptTransactionStatus.REJECTED, + Timestamp: now, + }; + const freshTerminal = { + ClReqID: 'c', + TransactionID: 'tx-c-corrected', + Status: ScryptTransactionStatus.COMPLETED, + TxHash: 'hash-corrected', + Timestamp: now, + }; + + (service as any).balanceTransactions.set('c', existingTerminal); + (service as any).cacheBalanceTransaction(freshTerminal); + + expect((service as any).balanceTransactions.get('c')).toBe(freshTerminal); + expect((service as any).balanceTransactions.get('c')).toEqual(freshTerminal); + }); + + it('cacheBalanceTransaction does not suppress keyless non-terminal after keyless terminal (ClReqID optional)', () => { + const now = new Date().toISOString(); + const terminalNoKey = { + TransactionID: 'tx-keyless-terminal', + Status: ScryptTransactionStatus.FAILED, + Timestamp: now, + }; + const nonTerminalNoKey = { + TransactionID: 'tx-keyless-nonterminal', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: now, + }; + + (service as any).cacheBalanceTransaction(terminalNoKey); + (service as any).cacheBalanceTransaction(nonTerminalNoKey); + + expect((service as any).balanceTransactions.get(undefined)).toEqual(nonTerminalNoKey); + }); + + it('registers catch-up via connection.onReconnect in the constructor', async () => { + expect(instance.onReconnect).toHaveBeenCalledTimes(1); + expect(instance.onReconnect).toHaveBeenCalledWith(expect.any(Function)); + + const registeredCallback = instance.onReconnect.mock.calls[0][0] as () => void | Promise; + instance.fetchAll.mockClear(); + + await registeredCallback(); + await flushPromises(); + + expect(instance.fetchAll).toHaveBeenCalledWith(ScryptMessageType.EXECUTION_REPORT); + expect(instance.fetchAll).toHaveBeenCalledWith(ScryptMessageType.BALANCE_TRANSACTION); + }); + + it('catchUpAfterReconnect applies the fulfilled stream even when the other stream rejects (Promise.allSettled isolation)', async () => { + const now = new Date().toISOString(); + const freshBalanceTx = { + ClReqID: 'iso-1', + TransactionID: 'tx-iso-new', + Status: ScryptTransactionStatus.COMPLETED, + TxHash: 'hash-iso', + Timestamp: now, + }; + const rejectionError = new Error('execution reports fetch failed'); + + const loggerErrorSpy = jest.spyOn((service as any).logger, 'error').mockImplementation(() => undefined); + + const registeredCallback = instance.onReconnect.mock.calls[0][0] as () => void | Promise; + instance.fetchAll.mockClear(); + instance.fetchAll.mockImplementation(async (streamName: string) => { + if (streamName === ScryptMessageType.EXECUTION_REPORT) throw rejectionError; + if (streamName === ScryptMessageType.BALANCE_TRANSACTION) return [freshBalanceTx]; + return []; + }); + + await registeredCallback(); + await flushPromises(); + + expect((service as any).balanceTransactions.get('iso-1')).toEqual(freshBalanceTx); + expect((service as any).executionReports.size).toBe(0); + expect(loggerErrorSpy).toHaveBeenCalledTimes(1); + expect(loggerErrorSpy).toHaveBeenCalledWith( + 'Scrypt reconnect catch-up (execution reports) failed:', + rejectionError, + ); + }); + + it('live BalanceTransaction subscriber goes through the terminal-aware guard', () => { + const now = new Date().toISOString(); + const existingTerminal = { + ClReqID: 'live-1', + TransactionID: 'tx-live-terminal', + Status: ScryptTransactionStatus.REJECTED, + Timestamp: now, + }; + (service as any).balanceTransactions.set('live-1', existingTerminal); + const terminalBefore = (service as any).balanceTransactions.get('live-1'); + + const balanceTxCall = instance.subscribeToStream.mock.calls.find( + ([streamName]) => streamName === ScryptMessageType.BALANCE_TRANSACTION, + ); + expect(balanceTxCall).toBeDefined(); + const liveSubscriber = balanceTxCall![1] as (transactions: unknown[]) => void; + + const nonTerminalUpdate = { + ClReqID: 'live-1', + TransactionID: 'tx-live-nonterminal', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: now, + }; + liveSubscriber([nonTerminalUpdate]); + + expect((service as any).balanceTransactions.get('live-1')).toBe(terminalBefore); + expect((service as any).balanceTransactions.get('live-1')).toEqual(existingTerminal); + }); + + it('live ExecutionReport subscriber goes through the terminal-aware guard', () => { + const now = new Date().toISOString(); + const existingTerminal = { + ClOrdID: 'ord-live-1', + Symbol: 'BTC-USD', + Side: 'Buy', + OrdStatus: ScryptOrderStatus.FILLED, + OrderQty: '1', + CumQty: '1', + LeavesQty: '0', + SubmitTime: now, + }; + (service as any).executionReports.set('ord-live-1', existingTerminal); + const terminalBefore = (service as any).executionReports.get('ord-live-1'); + + const executionReportCall = instance.subscribeToStream.mock.calls.find( + ([streamName]) => streamName === ScryptMessageType.EXECUTION_REPORT, + ); + expect(executionReportCall).toBeDefined(); + const liveSubscriber = executionReportCall![1] as (reports: unknown[]) => void; + + const nonTerminalUpdate = { + ClOrdID: 'ord-live-1', + Symbol: 'BTC-USD', + Side: 'Buy', + OrdStatus: ScryptOrderStatus.NEW, + OrderQty: '1', + CumQty: '0', + LeavesQty: '1', + SubmitTime: now, + }; + liveSubscriber([nonTerminalUpdate]); + + expect((service as any).executionReports.get('ord-live-1')).toBe(terminalBefore); + expect((service as any).executionReports.get('ord-live-1')).toEqual(existingTerminal); + }); + + it('getOrderStatus fallback does not clobber a live terminal push that arrived during the API await', async () => { + const now = new Date().toISOString(); + const staleNonTerminal = { + ClOrdID: 'X', + Symbol: 'BTC-USD', + Side: 'Buy', + OrdStatus: ScryptOrderStatus.NEW, + OrderQty: '1', + CumQty: '0', + LeavesQty: '1', + SubmitTime: now, + }; + const liveTerminal = { + ClOrdID: 'X', + Symbol: 'BTC-USD', + Side: 'Buy', + OrdStatus: ScryptOrderStatus.FILLED, + OrderQty: '1', + CumQty: '1', + LeavesQty: '0', + SubmitTime: now, + }; + + expect((service as any).executionReports.get('X')).toBeUndefined(); + + (instance as any).fetch.mockImplementation(async () => { + (service as any).cacheExecutionReport(liveTerminal); + return [staleNonTerminal]; + }); + + const result = await (service as any).getOrderStatus('X'); + + expect(result).toEqual(expect.objectContaining({ status: ScryptOrderStatus.FILLED })); + expect((service as any).executionReports.get('X')).toEqual(liveTerminal); + }); + + it('constructor warm-up BalanceTransaction fetch goes through the terminal-aware guard', async () => { + const now = new Date().toISOString(); + const terminalRecord = { + ClReqID: 'warm-1', + TransactionID: 'tx-warm-terminal', + Status: ScryptTransactionStatus.COMPLETED, + TxHash: 'hash-warm', + Timestamp: now, + }; + const nonTerminalDuplicate = { + ClReqID: 'warm-1', + TransactionID: 'tx-warm-nonterminal', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: now, + }; + + const MockedConnection = ScryptWebSocketConnection as jest.MockedClass; + MockedConnection.mockImplementationOnce( + () => + ({ + fetchAll: jest.fn().mockImplementation(async (streamName: string) => { + if (streamName === ScryptMessageType.BALANCE_TRANSACTION) { + return [terminalRecord, nonTerminalDuplicate]; + } + return []; + }), + fetch: jest.fn().mockResolvedValue([]), + subscribeToStream: jest.fn().mockReturnValue(() => undefined), + onReconnect: jest.fn(), + send: jest.fn(), + requestAndWaitForUpdate: jest.fn(), + }) as any, + ); + + const freshService = new ScryptService(); + await flushPromises(); + + const freshInstance = MockedConnection.mock.results[MockedConnection.mock.results.length - 1].value as any; + expect(freshInstance.fetchAll).toHaveBeenCalled(); + + expect((freshService as any).balanceTransactions.get('warm-1')).toEqual(terminalRecord); + expect((freshService as any).balanceTransactions.get('warm-1')).toBe(terminalRecord); + }); + + it('live BalanceTransaction without Timestamp is still cached (no age cutoff on live path)', () => { + const balanceTxCall = instance.subscribeToStream.mock.calls.find( + ([streamName]) => streamName === ScryptMessageType.BALANCE_TRANSACTION, + ); + expect(balanceTxCall).toBeDefined(); + const liveSubscriber = balanceTxCall![1] as (transactions: unknown[]) => void; + + const liveNoTimestamp = { + ClReqID: 'live-no-ts', + TransactionID: 'tx-live-no-ts', + Status: ScryptTransactionStatus.COMPLETED, + }; + liveSubscriber([liveNoTimestamp]); + + expect((service as any).balanceTransactions.get('live-no-ts')).toEqual(liveNoTimestamp); + }); + + it('live ExecutionReport older than 365 days is still cached (no age cutoff on live path)', () => { + const executionReportCall = instance.subscribeToStream.mock.calls.find( + ([streamName]) => streamName === ScryptMessageType.EXECUTION_REPORT, + ); + expect(executionReportCall).toBeDefined(); + const liveSubscriber = executionReportCall![1] as (reports: unknown[]) => void; + + const oldSubmitTime = new Date(Date.now() - 400 * 24 * 60 * 60 * 1000).toISOString(); + const oldReport = { + ClOrdID: 'ord-live-old', + Symbol: 'BTC-USD', + Side: 'Buy', + OrdStatus: ScryptOrderStatus.NEW, + OrderQty: '1', + CumQty: '0', + LeavesQty: '1', + SubmitTime: oldSubmitTime, + }; + liveSubscriber([oldReport]); + + expect((service as any).executionReports.get('ord-live-old')).toEqual(oldReport); + }); + + it('constructor warm-up drops BalanceTransaction older than 365 days (bulk age filter)', async () => { + const oldTimestamp = new Date(Date.now() - 400 * 24 * 60 * 60 * 1000).toISOString(); + const oldBalanceTx = { + ClReqID: 'warm-old', + TransactionID: 'tx-warm-old', + Status: ScryptTransactionStatus.COMPLETED, + Timestamp: oldTimestamp, + }; + + const MockedConnection = ScryptWebSocketConnection as jest.MockedClass; + MockedConnection.mockImplementationOnce( + () => + ({ + fetchAll: jest.fn().mockImplementation(async (streamName: string) => { + if (streamName === ScryptMessageType.BALANCE_TRANSACTION) { + return [oldBalanceTx]; + } + return []; + }), + fetch: jest.fn().mockResolvedValue([]), + subscribeToStream: jest.fn().mockReturnValue(() => undefined), + onReconnect: jest.fn(), + send: jest.fn(), + requestAndWaitForUpdate: jest.fn(), + }) as any, + ); + + const freshService = new ScryptService(); + await flushPromises(); + + expect((freshService as any).balanceTransactions.get('warm-old')).toBeUndefined(); + }); + + it('catchUpAfterReconnect coalesces overlapping reconnects into a second full run', async () => { + let resolveFirstExecutionReport: (value: unknown[]) => void; + const firstExecutionReportPromise = new Promise((resolve) => { + resolveFirstExecutionReport = resolve; + }); + let fetchAllCallCount = 0; + + instance.fetchAll.mockClear(); + instance.fetchAll.mockImplementation((streamName: string) => { + fetchAllCallCount += 1; + if (fetchAllCallCount === 1 && streamName === ScryptMessageType.EXECUTION_REPORT) { + return firstExecutionReportPromise; + } + return Promise.resolve([]); + }); + + const inFlight = (service as any).catchUpAfterReconnect(); + + // Overlapping reconnect while first catch-up is still pending on EXECUTION_REPORT + await (service as any).catchUpAfterReconnect(); + + resolveFirstExecutionReport!([]); + await inFlight; + await flushPromises(); + + const executionReportCalls = instance.fetchAll.mock.calls.filter( + ([streamName]) => streamName === ScryptMessageType.EXECUTION_REPORT, + ); + const balanceTransactionCalls = instance.fetchAll.mock.calls.filter( + ([streamName]) => streamName === ScryptMessageType.BALANCE_TRANSACTION, + ); + expect(executionReportCalls).toHaveLength(2); + expect(balanceTransactionCalls).toHaveLength(2); + }); +}); diff --git a/src/integration/exchange/services/scrypt-websocket-connection.ts b/src/integration/exchange/services/scrypt-websocket-connection.ts index f981995315..d93a6adb41 100644 --- a/src/integration/exchange/services/scrypt-websocket-connection.ts +++ b/src/integration/exchange/services/scrypt-websocket-connection.ts @@ -41,6 +41,7 @@ enum ScryptRequestType { SUBSCRIBE = 'subscribe', UNSUBSCRIBE = 'unsubscribe', PAGE = 'page', + CANCEL = 'cancel', } export const TRANSIENT_WS_ERROR_MARKERS = ['Connection closed', 'unknown reqid']; @@ -80,6 +81,7 @@ export class ScryptWebSocketConnection { 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; + private reconnectCallbacks: Array<() => void | Promise> = []; // requests private reqIdCounter = 0; @@ -103,14 +105,19 @@ export class ScryptWebSocketConnection { async fetch(streamName: ScryptMessageType, filters?: Record): Promise { const doFetch = async (): Promise => { - const response = await this.request({ - type: ScryptRequestType.SUBSCRIBE, - streams: [{ name: streamName, ...filters }], - }); + const reqId = ++this.reqIdCounter; + try { + const response = await this.requestWithId(reqId, { + type: ScryptRequestType.SUBSCRIBE, + streams: [{ name: streamName, ...filters }], + }); - if (!response.initial) throw new Error(`Expected initial ${streamName} message`); + if (!response.initial) throw new Error(`Expected initial ${streamName} message`); - return (response.data ?? []) as T[]; + return (response.data ?? []) as T[]; + } finally { + this.sendCancel(reqId); + } }; return this.retryOnTransientWsError(doFetch, `fetch ${streamName}`); @@ -121,32 +128,42 @@ export class ScryptWebSocketConnection { const allData: T[] = []; const reqId = ++this.reqIdCounter; - // First request - let response = await this.requestWithId(reqId, { - type: ScryptRequestType.SUBSCRIBE, - streams: [{ name: streamName, ...filters }], - }); + try { + // First request + let response = await this.requestWithId(reqId, { + type: ScryptRequestType.SUBSCRIBE, + streams: [{ name: streamName, ...filters }], + }); - if (!response.initial) throw new Error(`Expected initial ${streamName} message`); + if (!response.initial) throw new Error(`Expected initial ${streamName} message`); - allData.push(...((response.data ?? []) as T[])); + allData.push(...((response.data ?? []) as T[])); - // Paginate through all pages - while (response.next) { - response = await this.requestWithId(reqId, { - type: ScryptRequestType.PAGE, - streams: [{ name: streamName, after: response.next }], - }); + // Paginate through all pages + while (response.next) { + response = await this.requestWithId(reqId, { + type: ScryptRequestType.PAGE, + streams: [{ name: streamName, after: response.next }], + }); - allData.push(...((response.data ?? []) as T[])); - } + allData.push(...((response.data ?? []) as T[])); + } - return allData; + return allData; + } finally { + this.sendCancel(reqId); + } }; return this.retryOnTransientWsError(doFetch, `fetchAll ${streamName}`); } + // Register a callback fired after a successful RE-connect (not the first connect). Used to re-fetch state that + // a bare re-subscribe does not replay (see ScryptService catch-up). Callbacks must not throw / handle their own errors. + onReconnect(callback: () => void | Promise): void { + this.reconnectCallbacks.push(callback); + } + private async retryOnTransientWsError(operation: () => Promise, label: string): Promise { try { return await operation(); @@ -244,6 +261,7 @@ export class ScryptWebSocketConnection { // 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 { + const isReconnect = this.hasEverConnected; // 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); @@ -265,6 +283,17 @@ export class ScryptWebSocketConnection { clearTimeout(this.reconnectTimer); this.reconnectTimer = undefined; } + if (isReconnect) this.fireReconnectCallbacks(); + } + + private fireReconnectCallbacks(): void { + for (const cb of this.reconnectCallbacks) { + try { + void Promise.resolve(cb()).catch((e) => this.logger.error('Scrypt onReconnect callback rejected:', e)); + } catch (e) { + this.logger.error('Scrypt onReconnect callback failed:', e); + } + } } private assertCurrentGeneration(generation: number): void { @@ -436,6 +465,17 @@ export class ScryptWebSocketConnection { } } + // Stop an ad-hoc fetch/fetchAll stream by its request id (venue: cancel-by-reqid, no response). Best-effort: + // if the socket is not open the server-side stream is moot anyway; never throw (must not affect the fetch result). + private sendCancel(reqId: number): void { + if (!this.ws || this.ws.readyState !== WebSocket.OPEN) return; + try { + this.ws.send(JSON.stringify({ reqid: reqId, type: ScryptRequestType.CANCEL })); + } catch (e) { + this.logger.error(`Failed to cancel Scrypt stream ${reqId}:`, e); + } + } + // --- STREAMING SUBSCRIPTIONS --- // /** diff --git a/src/integration/exchange/services/scrypt.service.ts b/src/integration/exchange/services/scrypt.service.ts index f8a96b5dda..298d54ee05 100644 --- a/src/integration/exchange/services/scrypt.service.ts +++ b/src/integration/exchange/services/scrypt.service.ts @@ -38,6 +38,8 @@ export class ScryptService extends PricingProvider { private readonly balances?: AsyncSubscription>; private readonly executionReports: Map = new Map(); private readonly balanceTransactions: Map = new Map(); + private catchUpInProgress = false; + private catchUpPending = false; readonly name: string = 'Scrypt'; @@ -73,40 +75,102 @@ export class ScryptService extends PricingProvider { }); }); - const cacheMaxAge = Util.daysBefore(365); - // ExecutionReport subscription (all pages + subscription) this.connection .fetchAll(ScryptMessageType.EXECUTION_REPORT) - .then((reports) => { - const recent = reports.filter((r) => !r.SubmitTime || new Date(r.SubmitTime) >= cacheMaxAge); - for (const r of recent) this.executionReports.set(r.ClOrdID, r); - }) + .then((reports) => this.applyExecutionReports(reports)) .catch((error) => this.logger.error('Failed to fetch execution reports:', error)); this.connection.subscribeToStream(ScryptMessageType.EXECUTION_REPORT, (reports) => { - for (const report of reports) { - this.executionReports.set(report.ClOrdID, report); - } + for (const r of reports) this.cacheExecutionReport(r); // live event: always cache (terminal guard only, no age cutoff) }); // BalanceTransaction subscription (all pages + subscription) this.connection .fetchAll(ScryptMessageType.BALANCE_TRANSACTION) - .then((transactions) => { - const recent = transactions.filter((t) => new Date(t.Timestamp) >= cacheMaxAge); - for (const t of recent) this.balanceTransactions.set(t.ClReqID, t); - }) + .then((transactions) => this.applyBalanceTransactions(transactions)) .catch((error) => this.logger.error('Failed to fetch balance transactions:', error)); this.connection.subscribeToStream( ScryptMessageType.BALANCE_TRANSACTION, (transactions) => { - for (const t of transactions) { - this.balanceTransactions.set(t.ClReqID, t); - } + for (const t of transactions) this.cacheBalanceTransaction(t); // live event: always cache (terminal guard only) }, ); + + this.connection.onReconnect(() => this.catchUpAfterReconnect()); + } + + private isTerminalBalanceTransaction(t: ScryptBalanceTransaction): boolean { + return ( + [ScryptTransactionStatus.FAILED, ScryptTransactionStatus.REJECTED].includes(t.Status) || + (t.Status === ScryptTransactionStatus.COMPLETED && !!t.TxHash) + ); + } + + private cacheBalanceTransaction(t: ScryptBalanceTransaction): void { + const existing = this.balanceTransactions.get(t.ClReqID); + // Only apply the terminal guard for a real key: two distinct records that both lack a ClReqID collide under the + // `undefined` key, so the guard must not suppress one for the other (fall back to last-write-wins as before). + if (t.ClReqID && existing && this.isTerminalBalanceTransaction(existing) && !this.isTerminalBalanceTransaction(t)) + return; + this.balanceTransactions.set(t.ClReqID, t); + } + + private isTerminalExecutionReport(r: ScryptExecutionReport): boolean { + return [ScryptOrderStatus.FILLED, ScryptOrderStatus.CANCELED, ScryptOrderStatus.REJECTED].includes(r.OrdStatus); + } + + private cacheExecutionReport(r: ScryptExecutionReport): void { + const existing = this.executionReports.get(r.ClOrdID); + if (existing && this.isTerminalExecutionReport(existing) && !this.isTerminalExecutionReport(r)) return; + this.executionReports.set(r.ClOrdID, r); + } + + // Bulk (age-bounded) warm-up/catch-up path only — live subscriptions must cache directly via cacheExecutionReport/cacheBalanceTransaction, see constructor. + private applyExecutionReports(reports: ScryptExecutionReport[]): void { + const cacheMaxAge = Util.daysBefore(365); + for (const r of reports) if (!r.SubmitTime || new Date(r.SubmitTime) >= cacheMaxAge) this.cacheExecutionReport(r); + } + + // Bulk (age-bounded) warm-up/catch-up path only — live subscriptions must cache directly via cacheExecutionReport/cacheBalanceTransaction, see constructor. + private applyBalanceTransactions(transactions: ScryptBalanceTransaction[]): void { + const cacheMaxAge = Util.daysBefore(365); + for (const t of transactions) if (new Date(t.Timestamp) >= cacheMaxAge) this.cacheBalanceTransaction(t); + } + + // After a WS reconnect, re-fetch balance transactions + execution reports so an event missed during the outage + // is recovered (a bare re-subscribe is not documented to replay it). Mirrors the constructor warm-up's fetchAll, + // but not its subscribeToStream (resubscription is handled by the reconnect itself). Best-effort; logs on failure. + private async catchUpAfterReconnect(): Promise { + if (this.catchUpInProgress) { + this.catchUpPending = true; // coalesce: re-run once after the in-flight catch-up, to cover this reconnect's downtime + return; + } + this.catchUpInProgress = true; + try { + do { + this.catchUpPending = false; + const [executionResult, balanceResult] = await Promise.allSettled([ + this.connection.fetchAll(ScryptMessageType.EXECUTION_REPORT), + this.connection.fetchAll(ScryptMessageType.BALANCE_TRANSACTION), + ]); + + if (executionResult.status === 'fulfilled') { + this.applyExecutionReports(executionResult.value); + } else { + this.logger.error('Scrypt reconnect catch-up (execution reports) failed:', executionResult.reason); + } + + if (balanceResult.status === 'fulfilled') { + this.applyBalanceTransactions(balanceResult.value); + } else { + this.logger.error('Scrypt reconnect catch-up (balance transactions) failed:', balanceResult.reason); + } + } while (this.catchUpPending); + } finally { + this.catchUpInProgress = false; + } } // --- BALANCES --- // @@ -320,10 +384,13 @@ export class ScryptService extends PricingProvider { // Fallback: fetch from Scrypt API (e.g. after restart or WS reconnect) if (!report) { const reports = await this.fetchExecutionReports(Util.daysBefore(30)); - report = reports.find((r) => r.ClOrdID === clOrdId); + const fetched = reports.find((r) => r.ClOrdID === clOrdId); - if (report) { - this.executionReports.set(report.ClOrdID, report); + if (fetched) { + // Route through the terminal-aware guard: a live terminal push that arrived during the await + // must not be clobbered by a stale non-terminal snapshot from this fallback fetch. + this.cacheExecutionReport(fetched); + report = this.executionReports.get(clOrdId); } } From b5b0e3d0eb339a1d4ae2213f77a9228d4d09dc62 Mon Sep 17 00:00:00 2001 From: TaprootFreak <142087526+TaprootFreak@users.noreply.github.com> Date: Thu, 23 Jul 2026 21:37:22 +0200 Subject: [PATCH 2/2] fix(log): restore unfiltered pending netting (#4348) --- .../log/__tests__/log-job.service.spec.ts | 65 +++++++++++-------- .../supporting/log/log-job.service.ts | 45 ++----------- 2 files changed, 41 insertions(+), 69 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 bde4820a3d..b641ccbdec 100644 --- a/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts +++ b/src/subdomains/supporting/log/__tests__/log-job.service.spec.ts @@ -1831,12 +1831,12 @@ describe('LogJobService', () => { }); }); - describe('useUnfilteredTx per-leg clamp (toKrakenUnfiltered < 0)', () => { + describe('useUnfilteredTx nets unfiltered Kraken legs before aggregate clamp', () => { beforeEach(() => { (BankService as unknown as { ibanCache: Map }).ibanCache.clear(); (BankService as unknown as { ibanCache: Map }).ibanCache.set( - `${IbanBankName.YAPEAL}-CHF`, - yapealCHF.iban, + `${IbanBankName.YAPEAL}-EUR`, + yapealEUR.iban, ); }); @@ -1844,7 +1844,7 @@ describe('LogJobService', () => { (BankService as unknown as { ibanCache: Map }).ibanCache.clear(); }); - function setupUnfilteredToKrakenClamp(depositTx: ReturnType) { + function setupUnfilteredKrakenNetting(exchangeTxs: ReturnType[]): void { jest.spyOn(settingService, 'getCustomBalanceSettings').mockResolvedValue({ assets: [], addresses: [] }); jest.spyOn(settingService, 'getObj').mockImplementation(async (key, defaultValue) => { if (key === 'financeLogUnfilteredTx') return true as never; @@ -1879,45 +1879,54 @@ describe('LogJobService', () => { jest.spyOn(bankTxService, 'getRecentExchangeTx').mockResolvedValue([]); jest .spyOn(exchangeTxService, 'getRecentExchangeTx') - .mockImplementation(async (_minId, exchange, _types) => (exchange === ExchangeName.KRAKEN ? [depositTx] : [])); + .mockImplementation(async (_minId, exchange, _types) => (exchange === ExchangeName.KRAKEN ? exchangeTxs : [])); } - it('floors a negative unfiltered toKraken leg so plusBalance.total is 0 (not -10000)', async () => { - const yapealChfAsset = createCustomAsset({ - id: 8003, + it('nets opposing unfiltered Kraken legs so net-negative pending floors to 0 (prod regression)', async () => { + // Active Yapeal/EUR custody asset (sellable keeps it in the asset-log reduce). + const yapealEurAsset = createCustomAsset({ + id: 8004, blockchain: Blockchain.YAPEAL, - dexName: 'CHF', + dexName: 'EUR', sellable: true, }); - // Unmatched Kraken DEPOSIT credited to Yapeal CHF BIC → pendingBankAmount = -amount for toKrakenUnfiltered - const theDepositTx = createCustomExchangeTx({ + // Prod-shaped components for the same asset: + // fromKrakenUnfiltered = +881163.51 (unmatched Kraken WITHDRAWAL to bank) + // toKrakenUnfiltered = -5878500 (unmatched Kraken DEPOSIT from bank) + // Net totalPlusPending = -4997336.49 → aggregate clamp to 0. + // Per-leg clamps would zero only the negative leg and leave phantom pending +881163.51. + const fromKrakenWithdrawalTx = createCustomExchangeTx({ id: 5001, + type: ExchangeTxType.WITHDRAWAL, + currency: 'EUR', + method: 'Bank Frick (SEPA) International', + address: 'YAPEAL AG', + amount: 881163.51, + }); + const toKrakenDepositTx = createCustomExchangeTx({ + id: 5002, type: ExchangeTxType.DEPOSIT, status: 'ok', - currency: 'CHF', - method: 'Bank Frick (SIC) International', - address: yapealCHF.bic.padEnd(11, 'XXX'), - amount: 10000, + currency: 'EUR', + method: 'Bank Frick (SEPA) International', + address: yapealEUR.bic.padEnd(11, 'XXX'), + amount: 5878500, }); - setupUnfilteredToKrakenClamp(theDepositTx); + setupUnfilteredKrakenNetting([fromKrakenWithdrawalTx, toKrakenDepositTx]); const verboseSpy = jest.spyOn(service['logger'], 'verbose'); - const assetLog = await service['getAssetLog']([yapealChfAsset]); + const assetLog = await service['getAssetLog']([yapealEurAsset]); - // 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); + // Correct: opposing legs net first, then aggregate clamp floors the net-negative sum. + expect(assetLog[yapealEurAsset.id].plusBalance.total).toBe(0); + expect(assetLog[yapealEurAsset.id].plusBalance.pending?.fromKraken ?? 0).toBe(0); + expect(assetLog[yapealEurAsset.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); + // Aggregate clamp must have run (proves netting reached a negative totalPlusPending). + // Per-leg unfiltered clamps must not exist — they would leave total = 881163.51 and skip this path. + expect(verboseSpy.mock.calls.some((call) => String(call[0]).includes('totalPlusPending < 0'))).toBe(true); }); }); }); diff --git a/src/subdomains/supporting/log/log-job.service.ts b/src/subdomains/supporting/log/log-job.service.ts index 2cd855c8fd..77867dddd8 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; - let fromKrakenUnfiltered = + const fromKrakenUnfiltered = pendingChfKrakenYapealPlusAmountUnfiltered + pendingEurKrakenYapealPlusAmountUnfiltered + pendingKrakenYapealMinusAmountUnfiltered; - let toKrakenUnfiltered = + const toKrakenUnfiltered = pendingYapealKrakenPlusAmountUnfiltered + pendingChfYapealKrakenMinusAmountUnfiltered + pendingEurYapealKrakenMinusAmountUnfiltered; @@ -825,11 +825,11 @@ export class LogJobService { let fromScrypt = pendingChfScryptBankPlusAmount + pendingEurScryptBankPlusAmount + pendingScryptBankMinusAmount; let toScrypt = pendingBankScryptPlusAmount + pendingChfBankScryptMinusAmount + pendingEurBankScryptMinusAmount; - let fromScryptUnfiltered = + const fromScryptUnfiltered = pendingChfScryptBankPlusAmountUnfiltered + pendingEurScryptBankPlusAmountUnfiltered + pendingScryptBankMinusAmountUnfiltered; - let toScryptUnfiltered = + const toScryptUnfiltered = pendingBankScryptPlusAmountUnfiltered + pendingChfBankScryptMinusAmountUnfiltered + pendingEurBankScryptMinusAmountUnfiltered; @@ -901,43 +901,6 @@ 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 +