diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts new file mode 100644 index 0000000000..1526bf453b --- /dev/null +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -0,0 +1,444 @@ +import WebSocket from 'ws'; +import { LightningWebSocketClient } from '../lightning-ws-client'; + +const mockSockets: any[] = []; +let mockConstructorError: Error | undefined; + +jest.mock('ws', () => ({ + __esModule: true, + default: jest.fn().mockImplementation(() => { + if (mockConstructorError) throw mockConstructorError; + + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { EventEmitter } = require('events'); + + const socket = new EventEmitter(); + socket.send = jest.fn(); + socket.pong = jest.fn(); + mockSockets.push(socket); + + return socket; + }), +})); + +describe('LightningWebSocketClient', () => { + const RETRIES_PER_ALERT = 30; + const RETRY_WAIT_MS = 10000; + const MAX_DISCONNECTED_MS = 300000; + const MAX_REPORTED_CAUSES = 5; + const ALERT_WINDOW_MS = 3600000; + const URL = 'wss://test/v1/invoices/subscribe'; + + let client: LightningWebSocketClient; + let logger: { info: jest.SpyInstance; error: jest.SpyInstance }; + + beforeEach(() => { + jest.useFakeTimers(); + jest.clearAllMocks(); + mockSockets.length = 0; + mockConstructorError = undefined; + + client = new LightningWebSocketClient(URL, 'TestMacaroon'); + logger = { + info: jest.spyOn((client as any).logger, 'info').mockImplementation(() => undefined), + error: jest.spyOn((client as any).logger, 'error').mockImplementation(() => undefined), + }; + + client.setup({ add_index: '0' }); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('should connect with the macaroon header and a handshake timeout', () => { + expect(WebSocket).toHaveBeenCalledWith( + URL, + expect.objectContaining({ + headers: { 'Grpc-Metadata-Macaroon': 'TestMacaroon' }, + handshakeTimeout: 15000, + }), + ); + }); + + it('should reconnect after a close', () => { + reconnect(); + + expect(mockSockets).toHaveLength(2); + }); + + // Giving up permanently used to be survivable only because the process died and came back. + it('should keep retrying past the alert threshold', () => { + for (let i = 0; i < RETRIES_PER_ALERT * 2; i++) reconnect(); + + expect(mockSockets).toHaveLength(RETRIES_PER_ALERT * 2 + 1); + }); + + // --- signal: how long the stream has been down --- // + + // Immune to how fast each attempt fails. A slow-failing connect reconnects far too rarely to + // register as a rate, and such a stream is completely dead. + it('should report a stream that cannot connect however slowly it fails', () => { + for (let i = 0; i < 10; i++) { + jest.advanceTimersByTime(130000); + reconnect(); + } + + expect(errorsMatching(/no connection for/).length).toBeGreaterThan(0); + }); + + // The reported figure is the outage so far, not the configured threshold: a five-minute blip and + // a multi-hour outage must not read identically. + it('should report how long the stream has actually been down', () => { + for (let i = 0; i < 4; i++) { + jest.advanceTimersByTime(MAX_DISCONNECTED_MS); + reconnect(); + } + + const reported = errorsMatching(/no connection for (\d+) sec\./).map((m) => Number(m[1])); + + expect(reported.length).toBeGreaterThan(1); + expect(reported[1]).toBeGreaterThan(reported[0]); + }); + + it('should report downtime at most once per threshold', () => { + // 90 reconnects at 10 sec. apart span 890 sec., crossing the 300 sec. threshold twice + for (let i = 0; i < 90; i++) reconnect(); + + expect(errorsMatching(/no connection for/)).toHaveLength(2); + }); + + it('should stop reporting downtime once the stream connects', () => { + close(); + jest.advanceTimersByTime(MAX_DISCONNECTED_MS + RETRY_WAIT_MS); + open(); + logger.error.mockClear(); + + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(errorsMatching(/no connection for/)).toHaveLength(0); + }); + + // The figure has to come from the current outage, not from the last time the stream was ever up. + it('should measure a second outage from its own start', () => { + for (let i = 0; i < 31; i++) reconnect(); + open(); + jest.advanceTimersByTime(ALERT_WINDOW_MS); + logger.error.mockClear(); + + for (let i = 0; i < 31; i++) reconnect(); + + expect(errorsMatching(/no connection for (\d+) sec\./).map((m) => Number(m[1]))).toEqual([300]); + }); + + // --- signal: how often the stream reconnects --- // + + // Catches a stream that connects and drops straight back: downtime stays short, but LND sends no + // backlog on subscribe so every gap loses events. + it('should report a stream that reconnects constantly despite good cycles', () => { + for (let i = 0; i < RETRIES_PER_ALERT; i++) { + open(); + jest.advanceTimersByTime(60000); + reconnect(); + } + + // The span is the observed one, not the configured window — 30 reconnects in a minute and 30 + // spread over an hour must not read identically. + expect(logger.error).toHaveBeenCalledWith(expect.stringMatching(/30 reconnects within 2\d{3} sec\./)); + }); + + it('should not report a stream that reconnects rarely', () => { + for (let i = 0; i < RETRIES_PER_ALERT + 5; i++) { + open(); + jest.advanceTimersByTime(ALERT_WINDOW_MS); + reconnect(); + } + + expect(logger.error).not.toHaveBeenCalled(); + }); + + // Without the reset a permanently broken stream logs an error on every single reconnect. + it('should report reconnect bursts at most once per threshold', () => { + for (let i = 0; i < RETRIES_PER_ALERT * 2; i++) reconnect(); + + expect(errorsMatching(/reconnects within/)).toHaveLength(2); + }); + + // A wall clock that steps backwards must not resurrect reconnects that have aged out. + it('should measure the reconnect window on a monotonic clock', () => { + for (let i = 0; i < RETRIES_PER_ALERT - 1; i++) reconnect(); + + jest.advanceTimersByTime(ALERT_WINDOW_MS * 2); + jest.setSystemTime(new Date(Date.now() - 3 * 3600000)); + reconnect(); + + expect(errorsMatching(/reconnects within/)).toHaveLength(0); + }); + + // --- retry robustness --- // + + // Config reads the TLS certificate from disk per instantiation and throws when it is missing, so + // a cert rotation can make the reconnect throw. Uncaught inside the timer that would kill the + // process, and no socket means no further 'close' event to drive the next attempt. + it('should keep retrying when creating the socket throws', () => { + mockConstructorError = new Error('cert unavailable'); + + close(); + expect(() => jest.advanceTimersByTime(RETRY_WAIT_MS)).not.toThrow(); + expect(() => jest.advanceTimersByTime(RETRY_WAIT_MS * 5)).not.toThrow(); + + mockConstructorError = undefined; + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets).toHaveLength(2); + }); + + it('should log a repeating construction failure once', () => { + mockConstructorError = new Error('cert unavailable'); + + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS * 10); + + expect(errorsMatching(/retry failed/)).toHaveLength(1); + }); + + // Latching on a later 'open' would swallow this: construction can start succeeding while LND is + // still down, so 'open' never arrives to clear the latch. + it('should log a construction failure again after a successful construction', () => { + mockConstructorError = new Error('cert unavailable'); + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS * 3); + + mockConstructorError = undefined; + jest.advanceTimersByTime(RETRY_WAIT_MS); + + mockConstructorError = new Error('cert unavailable'); + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS * 3); + + expect(errorsMatching(/retry failed/)).toHaveLength(2); + }); + + it('should log a construction failure whose cause changes', () => { + mockConstructorError = new Error('cert unavailable'); + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS * 3); + + mockConstructorError = new Error('cert unreadable'); + jest.advanceTimersByTime(RETRY_WAIT_MS * 3); + + expect(errorsMatching(/retry failed/)).toHaveLength(2); + }); + + it('should bound construction failures that keep changing cause', () => { + close(); + + for (let i = 0; i < 40; i++) { + mockConstructorError = new Error(`cert unavailable ${i}`); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } + + expect(errorsMatching(/retry failed/)).toHaveLength(MAX_REPORTED_CAUSES); + }); + + // Retrying forever means this fires every 10 sec. for as long as LND is unreachable. + it('should log a connection error once per outage', () => { + for (let i = 0; i < 10; i++) { + latest().emit('error', new Error('ECONNREFUSED')); + reconnect(); + } + + expect(errorsMatching(/: error$/)).toHaveLength(1); + }); + + it('should log a connection error again after the stream recovers', () => { + latest().emit('error', new Error('ECONNREFUSED')); + reconnect(); + open(); + + latest().emit('error', new Error('ECONNREFUSED')); + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + // A crash-looping LND alternates between refused and rejected on every attempt. Remembering only + // the last cause would report every one of them, which is the per-attempt logging this avoids. + it('should bound connection errors that alternate cause', () => { + for (let i = 0; i < 50; i++) { + latest().emit('error', new Error(i % 2 ? 'connect ECONNREFUSED' : 'Unexpected server response: 502')); + reconnect(); + } + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + it('should bound connection errors that keep changing cause', () => { + for (let i = 0; i < 40; i++) { + latest().emit('error', new Error(`Unexpected server response: ${500 + i}`)); + reconnect(); + } + + expect(errorsMatching(/: error$/)).toHaveLength(MAX_REPORTED_CAUSES); + }); + + // Otherwise whichever causes arrived first read as the complete list, and a cause that needs a + // human — an expired macaroon behind a burst of gateway errors — never appears at all. + it('should say once that it is suppressing further causes', () => { + for (let i = 0; i < 40; i++) { + latest().emit('error', new Error(`Unexpected server response: ${500 + i}`)); + reconnect(); + } + + expect(errorsMatching(/further connection error causes suppressed/)).toHaveLength(1); + }); + + // Both cause sets share the reporting helper, so the notice has to say which log was truncated. + it('should name which log it is suppressing', () => { + close(); + + for (let i = 0; i < 40; i++) { + mockConstructorError = new Error(`cert unavailable ${i}`); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } + + expect(errorsMatching(/further retry failure causes suppressed/)).toHaveLength(1); + expect(errorsMatching(/further connection error causes suppressed/)).toHaveLength(0); + }); + + // read and write resets share a code and differ only in message, at an unchanging URL. + it('should tell apart errors that share a code but differ in message', () => { + const read: any = new Error('read ECONNRESET'); + read.code = 'ECONNRESET'; + const write: any = new Error('write ECONNRESET'); + write.code = 'ECONNRESET'; + + latest().emit('error', read); + reconnect(); + latest().emit('error', write); + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + // errorCause runs inside the retry catch: a throw there skips the reschedule and escapes the + // timer, which is a dead stream plus an uncaught exception. + it('should keep retrying when the construction error cannot be described', () => { + const hostile = Object.create(null); + hostile.self = hostile; + mockConstructorError = hostile; + + close(); + expect(() => jest.advanceTimersByTime(RETRY_WAIT_MS)).not.toThrow(); + + mockConstructorError = undefined; + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets).toHaveLength(2); + }); + + it('should tell apart errors that carry neither code nor message', () => { + latest().emit('error', 'boom'); + reconnect(); + latest().emit('error', 'bang'); + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + // A connect failure to a hostname arrives as an AggregateError with an empty message, so keying + // on the message alone would collapse every errno onto one entry. + it('should tell apart errors that differ only by code', () => { + const refused: any = new AggregateError([], ''); + refused.code = 'ECONNREFUSED'; + const unreachable: any = new AggregateError([], ''); + unreachable.code = 'EHOSTUNREACH'; + + latest().emit('error', refused); + reconnect(); + latest().emit('error', unreachable); + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + it('should survive an error it cannot describe', () => { + const hostile = Object.create(null); + hostile.self = hostile; + + expect(() => latest().emit('error', hostile)).not.toThrow(); + expect(errorsMatching(/: error$/)).toHaveLength(1); + }); + + // A rotated macaroon after a network outage never produces the 'open' that would clear a boolean + // latch, so the operator would keep seeing the stale cause and chase the wrong fault. + it('should log a connection error whose cause changes mid-outage', () => { + latest().emit('error', new Error('connect ECONNREFUSED')); + reconnect(); + latest().emit('error', new Error('Unexpected server response: 401')); + + expect(errorsMatching(/: error$/)).toHaveLength(2); + }); + + it('should not let a retired socket drive its successor', () => { + const retired = latest(); + + reconnect(); + retired.emit('ping', Buffer.from('ping')); + retired.emit('open'); + + expect(retired.pong).toHaveBeenCalledTimes(1); + expect(latest().pong).not.toHaveBeenCalled(); + expect(retired.send).toHaveBeenCalledTimes(1); + expect(latest().send).not.toHaveBeenCalled(); + }); + + // Without this the client connects and then says nothing, so LND sends nothing back and the + // socket stays open — no error, no close, and both health signals read it as healthy and quiet. + it('should send the subscribe request on open', () => { + open(); + + expect(latest().send).toHaveBeenCalledWith(JSON.stringify({ add_index: '0' })); + }); + + // --- messages --- // + + it('should emit parsed messages', () => { + const received: any[] = []; + client.subscribe((m) => received.push(m)); + + latest().emit('message', Buffer.from(JSON.stringify({ result: { state: 'OPEN' } }))); + + expect(received).toEqual([{ result: { state: 'OPEN' } }]); + }); + + it('should survive an unparsable message', () => { + const onError = jest.fn(); + client.subscribe({ next: () => undefined, error: onError }); + + expect(() => latest().emit('message', Buffer.from('not json'))).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + }); + + // --- HELPERS --- // + + function latest() { + return mockSockets[mockSockets.length - 1]; + } + + function open() { + latest().emit('open'); + } + + function close() { + latest().emit('close'); + } + + function reconnect() { + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } + + function errorsMatching(pattern: RegExp): RegExpMatchArray[] { + return logger.error.mock.calls + .map((c) => String(c[0]).match(pattern)) + .filter((m): m is RegExpMatchArray => m !== null); + } +}); diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 21808aab7d..0404e67389 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -4,14 +4,44 @@ import { GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; import WebSocket from 'ws'; +// Saturation is tracked beside the causes rather than as an entry among them, so no error message +// can be mistaken for the marker and quietly disable the notice. +interface ReportedCauses { + causes: Set; + saturated: boolean; +} + +// Runs inside the retry catch, so it must not throw: a throw there skips the reschedule and leaves +// the stream dead. `code` is part of the key because a connect failure to a hostname arrives as an +// AggregateError whose message is empty, which would otherwise collapse every errno onto one key. +function errorCause(e: any): string { + try { + const parts = [e?.code, e?.message].filter((p) => typeof p === 'string' && p.length); + + return parts.length ? parts.join(' ') : String(e); + } catch { + return ''; + } +} + export class LightningWebSocketClient extends Subject { private readonly logger = new LightningLogger(LightningWebSocketClient); private webSocket: WebSocket; - private readonly retryCounter = 30; private readonly retryWaitTimeSec = 10; - private retryAttempt = 0; + private readonly handshakeTimeoutSec = 15; + private readonly alertWindowSec = 3600; + private readonly retriesPerAlert = 30; + private readonly maxDisconnectedSec = 300; + + private readonly maxReportedCauses = 5; + + private retryTimestamps: number[] = []; + private disconnectedSince?: number; + private lastDowntimeAlert?: number; + private readonly connectErrors: ReportedCauses = { causes: new Set(), saturated: false }; + private readonly constructionErrors: ReportedCauses = { causes: new Set(), saturated: false }; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -30,6 +60,10 @@ export class LightningWebSocketClient extends Subject { ca: config.blockchain.lightning.certificate, }), + // Without this a stalled upgrade never resolves: ws emits neither 'error' nor 'close', so the + // retry chain has nothing to react to and the stream stops for good without reconnecting. + handshakeTimeout: this.handshakeTimeoutSec * 1000, + headers: { 'Grpc-Metadata-Macaroon': this.macaroon, }, @@ -37,42 +71,127 @@ export class LightningWebSocketClient extends Subject { } setup(openRequestBody: any) { - this.webSocket.on('open', () => { + // Bind the handlers to the socket they were registered on, not to the field, so a retired + // socket can never drive its successor. + const socket = this.webSocket; + + socket.on('open', () => { this.logger.info(`WebSocket ${this.wsUrl}: open`); - this.webSocket.send(JSON.stringify(openRequestBody)); + this.disconnectedSince = undefined; + this.lastDowntimeAlert = undefined; + this.clearReported(this.connectErrors); + + socket.send(JSON.stringify(openRequestBody)); }); - this.webSocket.on('error', (err: any) => { + socket.on('error', (err: any) => { + // Retrying forever means this fires every retryWaitTimeSec for as long as LND is unreachable. + // Report each distinct cause once per outage and let the throttled signals below carry the + // repetition. Remembering only the last cause would not do: a crash-looping LND alternates + // between refused and rejected on every attempt, which puts the per-attempt logging back. + if (!this.shouldReport(this.connectErrors, errorCause(err), 'connection error')) return; + this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); - this.webSocket.on('close', () => { + socket.on('close', () => { this.logger.info(`WebSocket ${this.wsUrl}: close`); - if (this.retryAttempt++ < this.retryCounter) { - setTimeout(() => { - this.logger.info(`WebSocket ${this.wsUrl}: retry ${this.retryAttempt}`); - this.createWebSocket(); - this.setup(openRequestBody); - }, this.retryWaitTimeSec * 1000); - } else { - this.logger.error(`WebSocket ${this.wsUrl}: closed after ${this.retryCounter} retries`); - } + this.scheduleRetry(openRequestBody); }); - this.webSocket.on('message', (message: string) => { + socket.on('message', (message: WebSocket.Data) => { try { - this.retryAttempt = 0; - - this.next(JSON.parse(message)); + this.next(JSON.parse(message.toString())); } catch (e) { this.logger.error(`WebSocket ${this.wsUrl}: Error during message processing`, e); } }); - this.webSocket.on('ping', (pingMessage: any) => { - this.webSocket.pong(pingMessage); + socket.on('ping', (pingMessage: any) => { + socket.pong(pingMessage); }); } + + private shouldReport(reported: ReportedCauses, cause: string, label: string): boolean { + if (reported.causes.has(cause)) return false; + + if (reported.causes.size >= this.maxReportedCauses) { + // Saturating silently would let whichever causes arrived first read as the whole diagnosis. + // The notice does not surface the dropped cause — an expired macaroon behind a burst of + // gateway errors is still not logged — it stops the list being read as complete. + if (!reported.saturated) { + reported.saturated = true; + this.logger.error(`WebSocket ${this.wsUrl}: further ${label} causes suppressed`); + } + + return false; + } + + reported.causes.add(cause); + + return true; + } + + private clearReported(reported: ReportedCauses) { + reported.causes.clear(); + reported.saturated = false; + } + + // Retry forever. The process used to die whenever a stream broke and came back with fresh + // connections; without that accidental safety net, giving up would leave ingestion dead for the + // lifetime of the process, with nothing erroring and nothing alerting. + // + // Two signals. How often the stream reconnects catches one that connects and drops straight back, + // which keeps the downtime short but delivers nothing, because LND sends no backlog on subscribe + // and every gap loses events for good. How long it has been down reports the size of an outage + // rather than just its existence, reaches a stream that cannot connect roughly twice as fast, and + // does not depend on handshakeTimeout keeping each attempt short enough to register as a rate. + private scheduleRetry(openRequestBody: any) { + // Monotonic: an NTP or DST correction must not fabricate or discard a window's worth of history. + const now = performance.now(); + + this.disconnectedSince ??= now; + this.lastDowntimeAlert ??= now; + + if (now - this.lastDowntimeAlert >= this.maxDisconnectedSec * 1000) { + const downSec = Math.round((now - this.disconnectedSince) / 1000); + this.logger.error(`WebSocket ${this.wsUrl}: no connection for ${downSec} sec.`); + + this.lastDowntimeAlert = now; + } + + this.retryTimestamps = this.retryTimestamps.filter((t) => now - t < this.alertWindowSec * 1000); + this.retryTimestamps.push(now); + + if (this.retryTimestamps.length >= this.retriesPerAlert) { + const spanSec = Math.round((now - this.retryTimestamps[0]) / 1000); + this.logger.error(`WebSocket ${this.wsUrl}: ${this.retryTimestamps.length} reconnects within ${spanSec} sec.`); + + this.retryTimestamps = []; + } + + setTimeout(() => { + this.logger.info(`WebSocket ${this.wsUrl}: retry`); + + try { + this.createWebSocket(); + this.setup(openRequestBody); + + this.clearReported(this.constructionErrors); + } catch (e) { + // Reading the TLS certificate happens per instantiation and throws on a missing file, so a + // cert rotation during a reconnect lands here. Uncaught it would kill the process, and with + // no socket created no 'close' event would ever drive the next attempt. Clear on a + // successful construction rather than on a later 'open', which never arrives while LND is + // down. + if (this.shouldReport(this.constructionErrors, errorCause(e), 'retry failure')) { + this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); + } + + this.scheduleRetry(openRequestBody); + } + }, this.retryWaitTimeSec * 1000); + } } diff --git a/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts new file mode 100644 index 0000000000..92c6c07b68 --- /dev/null +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -0,0 +1,270 @@ +import { Observable } from 'rxjs'; +import { Environment } from 'src/config/config'; +import { LightningWebSocketService } from '../lightning-ws.service'; + +const mockClients: any[] = []; + +jest.mock('../../lightning-ws-client', () => ({ + LightningWebSocketClient: jest.fn().mockImplementation(() => { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { Subject } = require('rxjs'); + + const client = new Subject(); + client.setup = jest.fn(); + mockClients.push(client); + + return client; + }), +})); + +describe('LightningWebSocketService', () => { + const ONCHAIN = 0; + const INVOICE = 1; + const PAYMENT = 2; + + let service: LightningWebSocketService; + + beforeAll(() => { + process.env.ENVIRONMENT = Environment.DEV; + process.env.LIGHTNING_LND_ADMIN_MACAROON = 'TestMacaroon'; + process.env.LIGHTNING_LND_WS_ONCHAIN_TRANSACTIONS_URL = 'wss://test/v1/transactions/subscribe'; + process.env.LIGHTNING_LND_WS_INVOICES_URL = 'wss://test/v1/invoices/subscribe'; + process.env.LIGHTNING_LND_WS_PAYMENTS_URL = 'wss://test/v2/router/payments'; + }); + + beforeEach(() => { + mockClients.length = 0; + service = new LightningWebSocketService(); + }); + + it('should open one WebSocket client per subscription', () => { + expect(mockClients).toHaveLength(3); + }); + + // LND's REST gateway ends a stream with an error frame that carries no `result`. That frame used + // to reach the mappers unbound, so `this.logger` threw and took the whole process down. + it('should survive a payment message without a result', () => { + const { received, onError } = collect(service.paymentTransactions); + + expect(() => + mockClients[PAYMENT].next({ error: { grpc_code: 1, http_code: 408, message: 'context canceled' } }), + ).not.toThrow(); + + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should survive an invoice message without a result', () => { + const { received, onError } = collect(service.invoiceTransactions); + + mockClients[INVOICE].next({ error: { message: 'context canceled' } }); + + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should survive an onchain message without a result', () => { + const { received, onError } = collect(service.onChainTransactions); + + mockClients[ONCHAIN].next({ error: { message: 'context canceled' } }); + + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + // `JSON.parse('null')` is a legal frame, and `null.result` throws in every mapper. Without these + // the onchain and payment pipes are never proved to contain a throw at all: their mappers cannot + // throw on any other input, so nothing else distinguishes them from an unguarded pipe. + it('should survive a null onchain frame', () => { + const { received, onError } = collect(service.onChainTransactions); + + expect(() => mockClients[ONCHAIN].next(null)).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should survive a null invoice frame', () => { + const { received, onError } = collect(service.invoiceTransactions); + + expect(() => mockClients[INVOICE].next(null)).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should survive a null payment frame', () => { + const { received, onError } = collect(service.paymentTransactions); + + expect(() => mockClients[PAYMENT].next(null)).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should survive an invoice message whose result cannot be mapped', () => { + const { received, onError } = collect(service.invoiceTransactions); + + // r_hash missing -> Buffer.from() throws inside the mapper + mockClients[INVOICE].next({ result: {} }); + + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + }); + + it('should log the offending frame as JSON', () => { + const logger = jest.spyOn((service as any).logger, 'error').mockImplementation(() => undefined); + + collect(service.paymentTransactions); + mockClients[PAYMENT].next({ error: { message: 'context canceled' } }); + + expect(logger).toHaveBeenCalledWith( + 'Result not available in payment message: {"error":{"message":"context canceled"}}', + ); + + logger.mockRestore(); + }); + + it('should log a frame that cannot be serialized', () => { + const logger = jest.spyOn((service as any).logger, 'error').mockImplementation(() => undefined); + const circular: any = { result: {} }; + circular.self = circular; + + const { onError } = collect(service.invoiceTransactions); + + expect(() => mockClients[INVOICE].next(circular)).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + expect(logger).toHaveBeenCalledWith(expect.stringContaining('[Circular'), expect.anything()); + + logger.mockRestore(); + }); + + // Both JSON.stringify and String() throw on this one, and describe() runs inside the handler that + // exists to keep a mapping failure from reaching RxJS. + it('should log a frame that cannot even be stringified', () => { + const logger = jest.spyOn((service as any).logger, 'error').mockImplementation(() => undefined); + const hostile: any = Object.create(null); + hostile.result = {}; + hostile.self = hostile; + + const { received, onError } = collect(service.invoiceTransactions); + + expect(() => mockClients[INVOICE].next(hostile)).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + expect(received).toEqual([undefined]); + expect(logger).toHaveBeenCalledWith(expect.stringContaining('[Object: null prototype]'), expect.anything()); + + logger.mockRestore(); + }); + + it('should keep delivering after a broken message', () => { + const { received, onError } = collect(service.paymentTransactions); + + mockClients[PAYMENT].next({ error: { message: 'context canceled' } }); + mockClients[PAYMENT].next(paymentMessage()); + + expect(onError).not.toHaveBeenCalled(); + expect(received).toHaveLength(2); + expect(received[1]?.transaction).toBe('TestPaymentHash'); + }); + + it('should map a payment message', () => { + const { received } = collect(service.paymentTransactions); + + mockClients[PAYMENT].next(paymentMessage()); + + expect(received[0]).toStrictEqual({ + state: 'SUCCEEDED', + transaction: 'TestPaymentHash', + secret: 'TestPreimage', + amount: -1000, + fee: -2, + creationTimestamp: new Date(1753790806000), + reason: 'FAILURE_REASON_NONE', + paymentRequest: 'lnbc1TestPaymentRequest', + }); + }); + + it('should map an invoice message', () => { + const { received } = collect(service.invoiceTransactions); + + mockClients[INVOICE].next(invoiceMessage('0')); + mockClients[INVOICE].next(invoiceMessage('1753790906')); + + expect(received[0]).toStrictEqual({ + state: 'OPEN', + transaction: '5465737448617368', + secret: '54657374507265696d616765', + amount: 1000, + fee: 0, + creationTimestamp: new Date(1753790806000), + expiresTimestamp: new Date(1753790806000 + 3600000), + confirmedTimestamp: undefined, + description: 'TestMemo', + paymentRequest: 'lnbc1TestPaymentRequest', + }); + + expect(received[1]?.confirmedTimestamp).toStrictEqual(new Date(1753790906000)); + }); + + it('should map an onchain message', () => { + const { received } = collect(service.onChainTransactions); + + mockClients[ONCHAIN].next({ + result: { + tx_hash: 'TestTxHash', + amount: '1000', + block_height: 951054, + time_stamp: '1753790806', + total_fees: '2', + }, + }); + + expect(received[0]).toStrictEqual({ + tx_hash: 'TestTxHash', + amount: '1000', + block_height: 951054, + time_stamp: '1753790806', + total_fees: '2', + }); + }); + + // --- HELPERS --- // + + function collect(observable: Observable) { + const received: T[] = []; + const onError = jest.fn(); + + observable.subscribe({ next: (m) => received.push(m), error: onError }); + + return { received, onError }; + } + + function invoiceMessage(settleDate: string) { + return { + result: { + state: 'OPEN', + r_hash: Buffer.from('TestHash').toString('base64'), + r_preimage: Buffer.from('TestPreimage').toString('base64'), + value: '1000', + creation_date: '1753790806', + expiry: '3600', + settle_date: settleDate, + memo: 'TestMemo', + payment_request: 'lnbc1TestPaymentRequest', + }, + }; + } + + function paymentMessage() { + return { + result: { + status: 'SUCCEEDED', + payment_hash: 'TestPaymentHash', + payment_preimage: 'TestPreimage', + value_sat: '1000', + fee_sat: '2', + creation_time_ns: '1753790806000000000', + failure_reason: 'FAILURE_REASON_NONE', + payment_request: 'lnbc1TestPaymentRequest', + }, + }; + } +}); diff --git a/src/integration/blockchain/lightning/services/lightning-ws.service.ts b/src/integration/blockchain/lightning/services/lightning-ws.service.ts index 6d2b7c3c66..c2d97872d7 100644 --- a/src/integration/blockchain/lightning/services/lightning-ws.service.ts +++ b/src/integration/blockchain/lightning/services/lightning-ws.service.ts @@ -2,6 +2,7 @@ import { Injectable } from '@nestjs/common'; import { EMPTY, Observable, map } from 'rxjs'; import { Configuration, Environment, GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; +import { inspect } from 'util'; import { LndOnchainTransactionDto, LndTransactionDto } from '../dto/lnd.dto'; import { LightningWebSocketClient } from '../lightning-ws-client'; @@ -32,9 +33,15 @@ export class LightningWebSocketService { this.setupInvoiceWebSocketClient(config); this.setupPaymentWebSocketClient(config); - this.onChainTransactions = this.onchainWebSocketClient.asObservable().pipe(map(this.mapOnchainMessage)); - this.invoiceTransactions = this.invoiceWebSocketClient.asObservable().pipe(map(this.mapInvoiceMessage)); - this.paymentTransactions = this.paymentWebSocketClient.asObservable().pipe(map(this.mapPaymentMessage)); + this.onChainTransactions = this.onchainWebSocketClient + .asObservable() + .pipe(map((message) => this.safeMap(message, () => this.mapOnchainMessage(message)))); + this.invoiceTransactions = this.invoiceWebSocketClient + .asObservable() + .pipe(map((message) => this.safeMap(message, () => this.mapInvoiceMessage(message)))); + this.paymentTransactions = this.paymentWebSocketClient + .asObservable() + .pipe(map((message) => this.safeMap(message, () => this.mapPaymentMessage(message)))); } private setupOnchainWebSocketClient(config: Configuration) { @@ -80,6 +87,36 @@ export class LightningWebSocketService { } // --- Message Handling --- // + + // RxJS turns a throw inside the map projection into an error notification. That terminates the + // subscription and, because the subscribers register no error handler, reaches + // reportUnhandledError, which kills the process. Mapping failures must therefore stay local. + // The mapper takes no argument on purpose: a bare `this.mapXMessage` reference does not satisfy + // `() => T`, so the unbound call that caused the crash cannot be written here. + private safeMap(message: any, mapper: () => T | undefined): T | undefined { + try { + return mapper(); + } catch (e) { + this.logger.error(`Error mapping WebSocket message: ${this.describe(message)}`, e); + return undefined; + } + } + + // Runs inside the failure handler, so it must not be able to throw itself. JSON.stringify rejects + // circular structures and returns undefined for undefined; String() throws on a null-prototype + // object. inspect() handles both, and the last resort covers whatever it does not. + private describe(message: any): string { + try { + return JSON.stringify(message) ?? inspect(message); + } catch { + try { + return inspect(message); + } catch { + return ''; + } + } + } + private mapOnchainMessage(onchainMessage: any): LndOnchainTransactionDto | undefined { const result = onchainMessage.result; @@ -93,7 +130,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in onchain message: ${onchainMessage}`); + this.logger.error(`Result not available in onchain message: ${this.describe(onchainMessage)}`); } private mapInvoiceMessage(invoiceMessage: any): LndTransactionDto | undefined { @@ -114,7 +151,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in invoice message: ${invoiceMessage}`); + this.logger.error(`Result not available in invoice message: ${this.describe(invoiceMessage)}`); } private mapPaymentMessage(paymentMessage: any): LndTransactionDto | undefined { @@ -133,6 +170,6 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in payment message: ${paymentMessage}`); + this.logger.error(`Result not available in payment message: ${this.describe(paymentMessage)}`); } }