From 7ecc02367e2a7cef5258247d01601c2b2336abf5 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 12:52:53 -0300 Subject: [PATCH 01/10] fix(lightning): keep a bad LND WebSocket frame from killing the process MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The three LND WebSocket streams were piped through `map(this.mapXMessage)`, which passes the mappers unbound. RxJS then calls them with `this === undefined`. The happy path never touches `this`, so this went unnoticed. The `result`-missing path does: it logs via `this.logger`, which throws a TypeError. RxJS converts a throw inside a map projection into an error notification, the subscribers register no error handler, so it reaches `reportUnhandledError` and takes the process down — tearing down all three WebSocket connections with it. LND's REST gateway ends a stream with an error frame that carries no `result`, so a cancelled TrackPayments subscription was enough to trigger it. - bind the mappers via arrow functions - route them through `safeMap` so a mapping failure stays local and the stream survives instead of terminating - log the offending frame as JSON; the template literal rendered `[object Object]` --- .../__test__/lightning-ws-service.spec.ts | 160 ++++++++++++++++++ .../services/lightning-ws.service.ts | 31 +++- 2 files changed, 185 insertions(+), 6 deletions(-) create mode 100644 src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts 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..5ac5f09f0e --- /dev/null +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -0,0 +1,160 @@ +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]); + }); + + it('should survive a message that 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 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 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 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..55c7b28edf 100644 --- a/src/integration/blockchain/lightning/services/lightning-ws.service.ts +++ b/src/integration/blockchain/lightning/services/lightning-ws.service.ts @@ -32,9 +32,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, (m) => this.mapOnchainMessage(m)))); + this.invoiceTransactions = this.invoiceWebSocketClient + .asObservable() + .pipe(map((message) => this.safeMap(message, (m) => this.mapInvoiceMessage(m)))); + this.paymentTransactions = this.paymentWebSocketClient + .asObservable() + .pipe(map((message) => this.safeMap(message, (m) => this.mapPaymentMessage(m)))); } private setupOnchainWebSocketClient(config: Configuration) { @@ -80,6 +86,19 @@ 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. + private safeMap(message: any, mapper: (message: any) => T | undefined): T | undefined { + try { + return mapper(message); + } catch (e) { + this.logger.error(`Error mapping WebSocket message: ${JSON.stringify(message)}`, e); + return undefined; + } + } + private mapOnchainMessage(onchainMessage: any): LndOnchainTransactionDto | undefined { const result = onchainMessage.result; @@ -93,7 +112,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in onchain message: ${onchainMessage}`); + this.logger.error(`Result not available in onchain message: ${JSON.stringify(onchainMessage)}`); } private mapInvoiceMessage(invoiceMessage: any): LndTransactionDto | undefined { @@ -114,7 +133,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in invoice message: ${invoiceMessage}`); + this.logger.error(`Result not available in invoice message: ${JSON.stringify(invoiceMessage)}`); } private mapPaymentMessage(paymentMessage: any): LndTransactionDto | undefined { @@ -133,6 +152,6 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in payment message: ${paymentMessage}`); + this.logger.error(`Result not available in payment message: ${JSON.stringify(paymentMessage)}`); } } From a40a3ffa1dcafe455f52e6a499d7aed1c25c0d31 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 13:07:56 -0300 Subject: [PATCH 02/10] fix(lightning): keep the WebSocket retry budget from draining silently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Surviving a bad frame instead of crashing only helps if the reconnect path can still recover. `retryAttempt` was reset solely when a message arrived, never when a connection opened, so the 30-strike budget drained monotonically for any stream that reconnects but stays quiet — LND sends no backlog on subscribe. Previously each such frame killed the process and the restart reset the counter; now the process survives, so the budget has to be restored on `open` or ingestion stops for good with nothing erroring and nothing alerting. Also address two ways the recovery paths could still take the process down: - `createWebSocket()` reads the TLS certificate per instantiation and throws on a missing file. In a bare `setTimeout` that throw is uncaught, and with no socket created no `close` event arrives to drive the next attempt. Retry now catches, logs and re-arms. - `JSON.stringify` was the one unguarded statement inside the mapping-failure handler, so an unserializable frame reproduced the crash from inside the guard meant to prevent it. `safeMap` now takes a zero-argument mapper: a bare `this.mapXMessage` reference does not satisfy `() => T`, so the unbound call this fix is about cannot be written at the call site any more. Tests: retry-budget restore, retry after a throwing socket construction, invoice mapping incl. both settle_date branches, and assertions on the logged frame. Verified all nine fail against the previous behaviour. --- .../__test__/lightning-ws-client.spec.ts | 118 ++++++++++++++++++ .../lightning/lightning-ws-client.ts | 40 ++++-- .../__test__/lightning-ws-service.spec.ts | 66 +++++++++- .../services/lightning-ws.service.ts | 28 +++-- 4 files changed, 231 insertions(+), 21 deletions(-) create mode 100644 src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts 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..3997601013 --- /dev/null +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -0,0 +1,118 @@ +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 RETRY_COUNTER = 30; + const RETRY_WAIT_MS = 10000; + + let client: LightningWebSocketClient; + + beforeEach(() => { + jest.useFakeTimers(); + mockSockets.length = 0; + mockConstructorError = undefined; + + client = new LightningWebSocketClient('wss://test/v1/invoices/subscribe', 'TestMacaroon'); + client.setup({ add_index: '0' }); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('should reconnect after a close', () => { + latest().emit('close'); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets).toHaveLength(2); + }); + + // The budget must measure consecutive failures to connect, not total reconnects. LND sends no + // backlog on subscribe, so a stream that reconnects but stays quiet would otherwise drain it and + // stop reconnecting for the lifetime of the process — silently, with nothing erroring. + it('should restore the retry budget once a connection opens', () => { + for (let i = 0; i <= RETRY_COUNTER; i++) { + latest().emit('close'); + jest.advanceTimersByTime(RETRY_WAIT_MS); + latest().emit('open'); + } + + const openedSockets = mockSockets.length; + + latest().emit('close'); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets.length).toBe(openedSockets + 1); + }); + + it('should give up after the retry budget is exhausted without a connection', () => { + for (let i = 0; i < RETRY_COUNTER; i++) { + latest().emit('close'); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } + + expect(mockSockets).toHaveLength(RETRY_COUNTER + 1); + + latest().emit('close'); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets).toHaveLength(RETRY_COUNTER + 1); + }); + + // 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'); + + latest().emit('close'); + expect(() => jest.advanceTimersByTime(RETRY_WAIT_MS)).not.toThrow(); + + mockConstructorError = undefined; + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(mockSockets).toHaveLength(2); + }); + + it('should emit parsed messages', () => { + const received: any[] = []; + client.subscribe((m) => received.push(m)); + + latest().emit('message', 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', 'not json')).not.toThrow(); + expect(onError).not.toHaveBeenCalled(); + }); + + // --- HELPERS --- // + + function latest() { + return mockSockets[mockSockets.length - 1]; + } +}); diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 21808aab7d..175b69fb0b 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -40,6 +40,11 @@ export class LightningWebSocketClient extends Subject { this.webSocket.on('open', () => { this.logger.info(`WebSocket ${this.wsUrl}: open`); + // The budget counts consecutive failures to establish a connection. Resetting it here — and + // not only once a message arrives — keeps a stream that reconnects but stays quiet (LND sends + // no backlog on subscribe) from draining it until reconnecting stops for good. + this.retryAttempt = 0; + this.webSocket.send(JSON.stringify(openRequestBody)); }); @@ -50,21 +55,11 @@ export class LightningWebSocketClient extends Subject { this.webSocket.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) => { try { - this.retryAttempt = 0; - this.next(JSON.parse(message)); } catch (e) { this.logger.error(`WebSocket ${this.wsUrl}: Error during message processing`, e); @@ -75,4 +70,27 @@ export class LightningWebSocketClient extends Subject { this.webSocket.pong(pingMessage); }); } + + private scheduleRetry(openRequestBody: any) { + if (this.retryAttempt++ >= this.retryCounter) { + this.logger.error(`WebSocket ${this.wsUrl}: closed after ${this.retryCounter} retries`); + return; + } + + setTimeout(() => { + this.logger.info(`WebSocket ${this.wsUrl}: retry ${this.retryAttempt}`); + + try { + this.createWebSocket(); + this.setup(openRequestBody); + } 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. + this.logger.error(`WebSocket ${this.wsUrl}: retry ${this.retryAttempt} 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 index 5ac5f09f0e..f17a2eb447 100644 --- a/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -72,7 +72,7 @@ describe('LightningWebSocketService', () => { expect(received).toEqual([undefined]); }); - it('should survive a message that cannot be mapped', () => { + 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 @@ -82,6 +82,32 @@ describe('LightningWebSocketService', () => { 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; + + collect(service.invoiceTransactions); + + expect(() => mockClients[INVOICE].next(circular)).not.toThrow(); + expect(logger).toHaveBeenCalledWith('Error mapping WebSocket message: [object Object]', expect.anything()); + + logger.mockRestore(); + }); + it('should keep delivering after a broken message', () => { const { received, onError } = collect(service.paymentTransactions); @@ -110,6 +136,28 @@ describe('LightningWebSocketService', () => { }); }); + 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); @@ -143,6 +191,22 @@ describe('LightningWebSocketService', () => { 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: { diff --git a/src/integration/blockchain/lightning/services/lightning-ws.service.ts b/src/integration/blockchain/lightning/services/lightning-ws.service.ts index 55c7b28edf..3c23d229fe 100644 --- a/src/integration/blockchain/lightning/services/lightning-ws.service.ts +++ b/src/integration/blockchain/lightning/services/lightning-ws.service.ts @@ -34,13 +34,13 @@ export class LightningWebSocketService { this.onChainTransactions = this.onchainWebSocketClient .asObservable() - .pipe(map((message) => this.safeMap(message, (m) => this.mapOnchainMessage(m)))); + .pipe(map((message) => this.safeMap(message, () => this.mapOnchainMessage(message)))); this.invoiceTransactions = this.invoiceWebSocketClient .asObservable() - .pipe(map((message) => this.safeMap(message, (m) => this.mapInvoiceMessage(m)))); + .pipe(map((message) => this.safeMap(message, () => this.mapInvoiceMessage(message)))); this.paymentTransactions = this.paymentWebSocketClient .asObservable() - .pipe(map((message) => this.safeMap(message, (m) => this.mapPaymentMessage(m)))); + .pipe(map((message) => this.safeMap(message, () => this.mapPaymentMessage(message)))); } private setupOnchainWebSocketClient(config: Configuration) { @@ -90,15 +90,25 @@ export class LightningWebSocketService { // 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. - private safeMap(message: any, mapper: (message: any) => T | undefined): T | undefined { + // 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(message); + return mapper(); } catch (e) { - this.logger.error(`Error mapping WebSocket message: ${JSON.stringify(message)}`, e); + this.logger.error(`Error mapping WebSocket message: ${this.describe(message)}`, e); return undefined; } } + private describe(message: any): string { + try { + return JSON.stringify(message); + } catch { + return String(message); + } + } + private mapOnchainMessage(onchainMessage: any): LndOnchainTransactionDto | undefined { const result = onchainMessage.result; @@ -112,7 +122,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in onchain message: ${JSON.stringify(onchainMessage)}`); + this.logger.error(`Result not available in onchain message: ${this.describe(onchainMessage)}`); } private mapInvoiceMessage(invoiceMessage: any): LndTransactionDto | undefined { @@ -133,7 +143,7 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in invoice message: ${JSON.stringify(invoiceMessage)}`); + this.logger.error(`Result not available in invoice message: ${this.describe(invoiceMessage)}`); } private mapPaymentMessage(paymentMessage: any): LndTransactionDto | undefined { @@ -152,6 +162,6 @@ export class LightningWebSocketService { }; } - this.logger.error(`Result not available in payment message: ${JSON.stringify(paymentMessage)}`); + this.logger.error(`Result not available in payment message: ${this.describe(paymentMessage)}`); } } From 8dc03c5f74b3e1a6de989d6890a5ce6e5a8ee22a Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 13:23:33 -0300 Subject: [PATCH 03/10] fix(lightning): never stop reconnecting, and make a stuck stream visible MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit reset the retry budget on `open`, which fixed the quiet stream but left both remaining paths ending in a silent, permanently dead subscription: - A connection that opens and drops straight back reset the budget every cycle, so it retried forever at info level and the only error the client can emit never fired. LND sends no backlog on subscribe, so such a stream delivers nothing at all. - A connection that never opens still gave up for good after 30 attempts. That was survivable before only because a broken stream killed the process and the restart built fresh connections; without that accidental safety net, an LND outage longer than five minutes ended ingestion until the next deploy. The budget now measures consecutive failures to hold a connection: it resets only after a connection has stood for a while, so a flap drains it like a failed connect. Retries no longer stop; instead every 30 attempts without a stable connection raise an error, so a stuck stream keeps trying and stays visible. Also: - register the socket handlers against the socket they belong to rather than the field, so a retired socket cannot drive its successor — the same aliasing class this branch set out to fix - make `describe()` total; `String()` throws on a null-prototype object, which would have escaped the very handler that keeps mapping failures off the stream - type the message handler as `WebSocket.Data` and parse via `toString()`; ws delivers a Buffer, not a string, and the spec now emits one Of the tests, 13 fail against the previous behaviour on develop and 5 against the preceding commit. The earlier claim that all nine of that commit's tests were discriminating was wrong: four were, the rest characterise existing behaviour. --- .../__test__/lightning-ws-client.spec.ts | 96 ++++++++++++++----- .../lightning/lightning-ws-client.ts | 50 ++++++---- .../__test__/lightning-ws-service.spec.ts | 23 ++++- .../services/lightning-ws.service.ts | 12 ++- 4 files changed, 135 insertions(+), 46 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index 3997601013..92efb5daf9 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -1,3 +1,4 @@ +import WebSocket from 'ws'; import { LightningWebSocketClient } from '../lightning-ws-client'; const mockSockets: any[] = []; @@ -21,17 +22,26 @@ jest.mock('ws', () => ({ })); describe('LightningWebSocketClient', () => { - const RETRY_COUNTER = 30; + const RETRIES_PER_ALERT = 30; const RETRY_WAIT_MS = 10000; + const MIN_CONNECTION_MS = 30000; + 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('wss://test/v1/invoices/subscribe', 'TestMacaroon'); + 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' }); }); @@ -39,43 +49,55 @@ describe('LightningWebSocketClient', () => { jest.useRealTimers(); }); + it('should connect with the macaroon header', () => { + expect(WebSocket).toHaveBeenCalledWith( + URL, + expect.objectContaining({ headers: { 'Grpc-Metadata-Macaroon': 'TestMacaroon' } }), + ); + }); + it('should reconnect after a close', () => { - latest().emit('close'); + close(); jest.advanceTimersByTime(RETRY_WAIT_MS); expect(mockSockets).toHaveLength(2); }); - // The budget must measure consecutive failures to connect, not total reconnects. LND sends no - // backlog on subscribe, so a stream that reconnects but stays quiet would otherwise drain it and - // stop reconnecting for the lifetime of the process — silently, with nothing erroring. - it('should restore the retry budget once a connection opens', () => { - for (let i = 0; i <= RETRY_COUNTER; i++) { - latest().emit('close'); + // LND sends no backlog on subscribe, so a healthy stream can stay silent indefinitely. The budget + // must measure consecutive failures to connect, not total reconnects. + it('should restore the retry budget once a connection holds', () => { + for (let i = 0; i <= RETRIES_PER_ALERT; i++) { + close(); jest.advanceTimersByTime(RETRY_WAIT_MS); - latest().emit('open'); + open(); + jest.advanceTimersByTime(MIN_CONNECTION_MS); } - const openedSockets = mockSockets.length; - - latest().emit('close'); - jest.advanceTimersByTime(RETRY_WAIT_MS); - - expect(mockSockets.length).toBe(openedSockets + 1); + expect(logger.error).not.toHaveBeenCalled(); }); - it('should give up after the retry budget is exhausted without a connection', () => { - for (let i = 0; i < RETRY_COUNTER; i++) { - latest().emit('close'); + // A connection that opens and drops straight back delivers nothing. Treating it as a success + // would reset the budget every cycle, so the flap would never raise an error. + it('should report a connection that never holds', () => { + for (let i = 0; i < RETRIES_PER_ALERT; i++) { + open(); + close(); jest.advanceTimersByTime(RETRY_WAIT_MS); } - expect(mockSockets).toHaveLength(RETRY_COUNTER + 1); + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 30 retries`); + }); - latest().emit('close'); - jest.advanceTimersByTime(RETRY_WAIT_MS); + // 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++) { + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } - expect(mockSockets).toHaveLength(RETRY_COUNTER + 1); + expect(mockSockets).toHaveLength(RETRIES_PER_ALERT * 2 + 1); + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 30 retries`); + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 60 retries`); }); // Config reads the TLS certificate from disk per instantiation and throws when it is missing, so @@ -84,8 +106,9 @@ describe('LightningWebSocketClient', () => { it('should keep retrying when creating the socket throws', () => { mockConstructorError = new Error('cert unavailable'); - latest().emit('close'); + 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); @@ -93,11 +116,24 @@ describe('LightningWebSocketClient', () => { expect(mockSockets).toHaveLength(2); }); + it('should not let a retired socket drive its successor', () => { + const retired = latest(); + + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + retired.emit('ping', Buffer.from('ping')); + + expect(retired.pong).toHaveBeenCalledTimes(1); + expect(latest().pong).not.toHaveBeenCalled(); + }); + it('should emit parsed messages', () => { const received: any[] = []; client.subscribe((m) => received.push(m)); - latest().emit('message', JSON.stringify({ result: { state: 'OPEN' } })); + // ws delivers RawData, not a string + latest().emit('message', Buffer.from(JSON.stringify({ result: { state: 'OPEN' } }))); expect(received).toEqual([{ result: { state: 'OPEN' } }]); }); @@ -106,7 +142,7 @@ describe('LightningWebSocketClient', () => { const onError = jest.fn(); client.subscribe({ next: () => undefined, error: onError }); - expect(() => latest().emit('message', 'not json')).not.toThrow(); + expect(() => latest().emit('message', Buffer.from('not json'))).not.toThrow(); expect(onError).not.toHaveBeenCalled(); }); @@ -115,4 +151,12 @@ describe('LightningWebSocketClient', () => { function latest() { return mockSockets[mockSockets.length - 1]; } + + function open() { + latest().emit('open'); + } + + function close() { + latest().emit('close'); + } }); diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 175b69fb0b..97b2fabb64 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -9,9 +9,12 @@ export class LightningWebSocketClient extends Subject { private webSocket: WebSocket; - private readonly retryCounter = 30; private readonly retryWaitTimeSec = 10; + private readonly retriesPerAlert = 30; + private readonly minConnectionTimeSec = 30; + private retryAttempt = 0; + private connectedAt?: number; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -37,44 +40,59 @@ 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`); - // The budget counts consecutive failures to establish a connection. Resetting it here — and - // not only once a message arrives — keeps a stream that reconnects but stays quiet (LND sends - // no backlog on subscribe) from draining it until reconnecting stops for good. - this.retryAttempt = 0; + this.connectedAt = Date.now(); - this.webSocket.send(JSON.stringify(openRequestBody)); + socket.send(JSON.stringify(openRequestBody)); }); - this.webSocket.on('error', (err: any) => { + socket.on('error', (err: any) => { this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); - this.webSocket.on('close', () => { + socket.on('close', () => { this.logger.info(`WebSocket ${this.wsUrl}: close`); + // Only a connection that actually held counts as a success. Resetting the budget on every + // 'open' would hide an endless flap: LND sends no backlog on subscribe, so a stream that + // connects and drops again delivers nothing and would otherwise never raise an error. + if (this.heldForSec() >= this.minConnectionTimeSec) this.retryAttempt = 0; + this.connectedAt = undefined; + this.scheduleRetry(openRequestBody); }); - this.webSocket.on('message', (message: string) => { + socket.on('message', (message: WebSocket.Data) => { try { - 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 heldForSec(): number { + return this.connectedAt ? (Date.now() - this.connectedAt) / 1000 : 0; + } + private scheduleRetry(openRequestBody: any) { - if (this.retryAttempt++ >= this.retryCounter) { - this.logger.error(`WebSocket ${this.wsUrl}: closed after ${this.retryCounter} retries`); - return; + this.retryAttempt++; + + // 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. + if (this.retryAttempt % this.retriesPerAlert === 0) { + this.logger.error(`WebSocket ${this.wsUrl}: no stable connection after ${this.retryAttempt} retries`); } setTimeout(() => { 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 index f17a2eb447..3da6e9c912 100644 --- a/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -100,10 +100,29 @@ describe('LightningWebSocketService', () => { const circular: any = { result: {} }; circular.self = circular; - collect(service.invoiceTransactions); + const { onError } = collect(service.invoiceTransactions); expect(() => mockClients[INVOICE].next(circular)).not.toThrow(); - expect(logger).toHaveBeenCalledWith('Error mapping WebSocket message: [object Object]', expect.anything()); + 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).toHaveBeenCalled(); logger.mockRestore(); }); diff --git a/src/integration/blockchain/lightning/services/lightning-ws.service.ts b/src/integration/blockchain/lightning/services/lightning-ws.service.ts index 3c23d229fe..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'; @@ -101,11 +102,18 @@ export class LightningWebSocketService { } } + // 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); + return JSON.stringify(message) ?? inspect(message); } catch { - return String(message); + try { + return inspect(message); + } catch { + return ''; + } } } From d1e95cf186eb3467f4104c951cbdd434c13d4453 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 13:40:01 -0300 Subject: [PATCH 04/10] fix(lightning): judge stream health by reconnect rate, not consecutive failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A budget of consecutive failures resets on any good cycle, so a single connection that holds hides everything around it. A stream alternating 28 short connections with one long one ran at 17% uptime without logging anything, and a steady 31s-up/10s-down loop — what an idle timeout in front of LND produces — reconnected 2107 times in a simulated day, also silently. Counting reconnects within a rolling hour has no such blind spot, and it covers both shapes the previous version needed separate handling for: a stream that cannot connect and one that connects and drops straight back both reconnect far more often than a healthy stream, which only does so on an LND restart or a deploy. That removes the hold-duration threshold, the consecutive counter and its off-by-one after a reset. Measured against the same simulations: mixed flap 580 reconnects -> 19 errors, steady loop 2107 -> 70, and a healthy stream reconnecting every two hours for a week stays silent. Also: assert the inspected output in the unserializable-frame test rather than only that logging happened, and drop a comment claiming a test covers the Buffer message path — JSON.parse already coerced a Buffer, so that change is type-only. --- .../__test__/lightning-ws-client.spec.ts | 61 +++++++++---------- .../lightning/lightning-ws-client.ts | 44 +++++++------ .../__test__/lightning-ws-service.spec.ts | 2 +- 3 files changed, 52 insertions(+), 55 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index 92efb5daf9..fafaf0d3fa 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -24,7 +24,7 @@ jest.mock('ws', () => ({ describe('LightningWebSocketClient', () => { const RETRIES_PER_ALERT = 30; const RETRY_WAIT_MS = 10000; - const MIN_CONNECTION_MS = 30000; + const ALERT_WINDOW_MS = 3600000; const URL = 'wss://test/v1/invoices/subscribe'; let client: LightningWebSocketClient; @@ -57,47 +57,44 @@ describe('LightningWebSocketClient', () => { }); it('should reconnect after a close', () => { - close(); - jest.advanceTimersByTime(RETRY_WAIT_MS); + reconnect(); expect(mockSockets).toHaveLength(2); }); - // LND sends no backlog on subscribe, so a healthy stream can stay silent indefinitely. The budget - // must measure consecutive failures to connect, not total reconnects. - it('should restore the retry budget once a connection holds', () => { - for (let i = 0; i <= RETRIES_PER_ALERT; i++) { - close(); - jest.advanceTimersByTime(RETRY_WAIT_MS); - open(); - jest.advanceTimersByTime(MIN_CONNECTION_MS); - } + // 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(logger.error).not.toHaveBeenCalled(); + expect(mockSockets).toHaveLength(RETRIES_PER_ALERT * 2 + 1); + }); + + it('should report a stream that cannot connect', () => { + for (let i = 0; i < RETRIES_PER_ALERT; i++) reconnect(); + + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: 30 reconnects within 3600 sec.`); }); - // A connection that opens and drops straight back delivers nothing. Treating it as a success - // would reset the budget every cycle, so the flap would never raise an error. - it('should report a connection that never holds', () => { + // The signal has to be a rate, not a run of consecutive failures: an occasional good cycle would + // reset a consecutive counter and hide a stream that is losing most of its events. + it('should report a stream that reconnects constantly despite occasional good cycles', () => { for (let i = 0; i < RETRIES_PER_ALERT; i++) { open(); - close(); - jest.advanceTimersByTime(RETRY_WAIT_MS); + jest.advanceTimersByTime(60000); + reconnect(); } - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 30 retries`); + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: 30 reconnects within 3600 sec.`); }); - // 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++) { - close(); - jest.advanceTimersByTime(RETRY_WAIT_MS); + it('should not report a stream that reconnects rarely', () => { + for (let i = 0; i < RETRIES_PER_ALERT * 3; i++) { + open(); + jest.advanceTimersByTime(ALERT_WINDOW_MS); + reconnect(); } - expect(mockSockets).toHaveLength(RETRIES_PER_ALERT * 2 + 1); - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 30 retries`); - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no stable connection after 60 retries`); + expect(logger.error).not.toHaveBeenCalled(); }); // Config reads the TLS certificate from disk per instantiation and throws when it is missing, so @@ -119,9 +116,7 @@ describe('LightningWebSocketClient', () => { it('should not let a retired socket drive its successor', () => { const retired = latest(); - close(); - jest.advanceTimersByTime(RETRY_WAIT_MS); - + reconnect(); retired.emit('ping', Buffer.from('ping')); expect(retired.pong).toHaveBeenCalledTimes(1); @@ -132,7 +127,6 @@ describe('LightningWebSocketClient', () => { const received: any[] = []; client.subscribe((m) => received.push(m)); - // ws delivers RawData, not a string latest().emit('message', Buffer.from(JSON.stringify({ result: { state: 'OPEN' } }))); expect(received).toEqual([{ result: { state: 'OPEN' } }]); @@ -159,4 +153,9 @@ describe('LightningWebSocketClient', () => { function close() { latest().emit('close'); } + + function reconnect() { + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + } }); diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 97b2fabb64..d9155d3a8c 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -10,11 +10,10 @@ export class LightningWebSocketClient extends Subject { private webSocket: WebSocket; private readonly retryWaitTimeSec = 10; + private readonly alertWindowSec = 3600; private readonly retriesPerAlert = 30; - private readonly minConnectionTimeSec = 30; - private retryAttempt = 0; - private connectedAt?: number; + private retryTimestamps: number[] = []; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -47,8 +46,6 @@ export class LightningWebSocketClient extends Subject { socket.on('open', () => { this.logger.info(`WebSocket ${this.wsUrl}: open`); - this.connectedAt = Date.now(); - socket.send(JSON.stringify(openRequestBody)); }); @@ -59,12 +56,6 @@ export class LightningWebSocketClient extends Subject { socket.on('close', () => { this.logger.info(`WebSocket ${this.wsUrl}: close`); - // Only a connection that actually held counts as a success. Resetting the budget on every - // 'open' would hide an endless flap: LND sends no backlog on subscribe, so a stream that - // connects and drops again delivers nothing and would otherwise never raise an error. - if (this.heldForSec() >= this.minConnectionTimeSec) this.retryAttempt = 0; - this.connectedAt = undefined; - this.scheduleRetry(openRequestBody); }); @@ -81,22 +72,29 @@ export class LightningWebSocketClient extends Subject { }); } - private heldForSec(): number { - return this.connectedAt ? (Date.now() - this.connectedAt) / 1000 : 0; - } - + // 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. + // + // How often a stream reconnects is what separates a healthy one — LND restarts, deploys — from + // one that cannot hold a connection, and unlike a run of consecutive failures it does not reset + // itself on the occasional successful cycle. LND sends no backlog on subscribe, so every gap + // drops events for good; a stream reconnecting this often is losing data either way. private scheduleRetry(openRequestBody: any) { - this.retryAttempt++; + const now = Date.now(); + this.retryTimestamps = this.retryTimestamps.filter((t) => now - t < this.alertWindowSec * 1000); + this.retryTimestamps.push(now); + + if (this.retryTimestamps.length >= this.retriesPerAlert) { + this.logger.error( + `WebSocket ${this.wsUrl}: ${this.retryTimestamps.length} reconnects within ${this.alertWindowSec} sec.`, + ); - // 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. - if (this.retryAttempt % this.retriesPerAlert === 0) { - this.logger.error(`WebSocket ${this.wsUrl}: no stable connection after ${this.retryAttempt} retries`); + this.retryTimestamps = []; } setTimeout(() => { - this.logger.info(`WebSocket ${this.wsUrl}: retry ${this.retryAttempt}`); + this.logger.info(`WebSocket ${this.wsUrl}: retry`); try { this.createWebSocket(); @@ -105,7 +103,7 @@ export class LightningWebSocketClient extends Subject { // 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. - this.logger.error(`WebSocket ${this.wsUrl}: retry ${this.retryAttempt} failed`, e); + this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); this.scheduleRetry(openRequestBody); } 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 index 3da6e9c912..34347c00d7 100644 --- a/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -122,7 +122,7 @@ describe('LightningWebSocketService', () => { expect(() => mockClients[INVOICE].next(hostile)).not.toThrow(); expect(onError).not.toHaveBeenCalled(); expect(received).toEqual([undefined]); - expect(logger).toHaveBeenCalled(); + expect(logger).toHaveBeenCalledWith(expect.stringContaining('[Object: null prototype]'), expect.anything()); logger.mockRestore(); }); From 65e4631fa81fac0a81a10b72d46b77be0109ced3 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 14:02:35 -0300 Subject: [PATCH 05/10] fix(lightning): detect the stream deaths a reconnect rate cannot see MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Judging health purely by reconnect rate has a floor: 30 reconnects per hour means any repeating failure cycle of 125 sec. or longer never registers. That band is occupied. No handshake timeout was set, so a connect attempt against a dropped (rather than refused) port fails at the TCP SYN timeout, ~127 sec., giving a 137 sec. cycle — 26 reconnects an hour on a stream that is completely dead, and the previous commit reported it as healthy. The consecutive counter it replaced had no such floor, so that was a regression within this branch. Worse, a rate cannot see a stream with no reconnects at all. Without a handshake timeout a stalled upgrade emits neither 'error' nor 'close', and a half-open connection never closes either, since ws disables the socket timeout and this client answers pings but never sends any. Both leave a stream delivering nothing while every signal reads it as healthy and quiet. - track how long the stream has been disconnected, reset on 'open'. This is immune to how slowly each attempt fails, and covers what the rate cannot - set handshakeTimeout, so a stalled upgrade becomes an 'error' and a 'close' that the retry chain can act on - ping on an interval and terminate when a pong does not come back, which turns a half-open connection into a reconnect - report the observed span in the reconnect alert rather than the configured window, which was constant by construction - log a repeating construction failure once instead of every 10 sec.; the signals above already carry the repetition - measure with the monotonic clock, so a backwards NTP or DST correction cannot fabricate a window's worth of reconnects Simulated over 24 h: SYN blackhole 630 reconnects, 0 alerts before, 209 now; half-open goes from silent forever to detected in 60 sec.; a healthy stream reconnecting every two hours for a week still logs nothing. 18 tests fail against develop and 5 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 127 ++++++++++++++++-- .../lightning/lightning-ws-client.ts | 77 +++++++++-- 2 files changed, 182 insertions(+), 22 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index fafaf0d3fa..09d41bdccd 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -14,7 +14,9 @@ jest.mock('ws', () => ({ const socket = new EventEmitter(); socket.send = jest.fn(); + socket.ping = jest.fn(); socket.pong = jest.fn(); + socket.terminate = jest.fn(() => socket.emit('close')); mockSockets.push(socket); return socket; @@ -24,6 +26,8 @@ jest.mock('ws', () => ({ describe('LightningWebSocketClient', () => { const RETRIES_PER_ALERT = 30; const RETRY_WAIT_MS = 10000; + const PING_INTERVAL_MS = 30000; + const MAX_DISCONNECTED_MS = 300000; const ALERT_WINDOW_MS = 3600000; const URL = 'wss://test/v1/invoices/subscribe'; @@ -49,10 +53,13 @@ describe('LightningWebSocketClient', () => { jest.useRealTimers(); }); - it('should connect with the macaroon header', () => { + it('should connect with the macaroon header and a handshake timeout', () => { expect(WebSocket).toHaveBeenCalledWith( URL, - expect.objectContaining({ headers: { 'Grpc-Metadata-Macaroon': 'TestMacaroon' } }), + expect.objectContaining({ + headers: { 'Grpc-Metadata-Macaroon': 'TestMacaroon' }, + handshakeTimeout: 15000, + }), ); }); @@ -69,34 +76,111 @@ describe('LightningWebSocketClient', () => { expect(mockSockets).toHaveLength(RETRIES_PER_ALERT * 2 + 1); }); - it('should report a stream that cannot connect', () => { - for (let i = 0; i < RETRIES_PER_ALERT; i++) reconnect(); + // --- signal: how long the stream has been down --- // - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: 30 reconnects within 3600 sec.`); + // Immune to how fast each attempt fails. A connect that times out after two minutes 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', () => { + const slowFailureMs = 130000; + + for (let i = 0; i < 10; i++) { + jest.advanceTimersByTime(slowFailureMs); + reconnect(); + } + + expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no connection for 300 sec.`); + }); + + it('should stop reporting downtime once the stream connects', () => { + close(); + jest.advanceTimersByTime(MAX_DISCONNECTED_MS); + jest.advanceTimersByTime(RETRY_WAIT_MS); + open(); + logger.error.mockClear(); + + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS); + + expect(logger.error).not.toHaveBeenCalledWith(expect.stringContaining('no connection for')); }); - // The signal has to be a rate, not a run of consecutive failures: an occasional good cycle would - // reset a consecutive counter and hide a stream that is losing most of its events. - it('should report a stream that reconnects constantly despite occasional good cycles', () => { + // --- 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); + hold(60000); reconnect(); } - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: 30 reconnects within 3600 sec.`); + // 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 * 3; i++) { + for (let i = 0; i < RETRIES_PER_ALERT + 5; i++) { open(); - jest.advanceTimersByTime(ALERT_WINDOW_MS); + hold(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(logger.error.mock.calls.filter((c) => /reconnects within/.test(c[0]))).toHaveLength(2); + }); + + // --- liveness --- // + + // A half-open connection yields no traffic and no 'close', so both signals above would read it as + // a healthy quiet stream forever. + it('should reconnect when pings go unanswered', () => { + open(); + const stalled = latest(); + + jest.advanceTimersByTime(PING_INTERVAL_MS); + expect(stalled.ping).toHaveBeenCalledTimes(1); + + jest.advanceTimersByTime(PING_INTERVAL_MS); + expect(stalled.terminate).toHaveBeenCalled(); + expect(logger.error).toHaveBeenCalledWith(expect.stringContaining('no pong')); + + jest.advanceTimersByTime(RETRY_WAIT_MS); + expect(mockSockets).toHaveLength(2); + }); + + it('should keep a connection that answers pings', () => { + open(); + const socket = latest(); + + for (let i = 0; i < 10; i++) { + jest.advanceTimersByTime(PING_INTERVAL_MS); + socket.emit('pong'); + } + + expect(socket.terminate).not.toHaveBeenCalled(); + expect(mockSockets).toHaveLength(1); + }); + + it('should stop pinging a closed connection', () => { + open(); + const socket = latest(); + close(); + + jest.advanceTimersByTime(PING_INTERVAL_MS * 3); + + expect(socket.ping).not.toHaveBeenCalled(); + }); + + // --- 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. @@ -113,6 +197,15 @@ describe('LightningWebSocketClient', () => { 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(logger.error.mock.calls.filter((c) => /retry failed/.test(c[0]))).toHaveLength(1); + }); + it('should not let a retired socket drive its successor', () => { const retired = latest(); @@ -123,6 +216,8 @@ describe('LightningWebSocketClient', () => { expect(latest().pong).not.toHaveBeenCalled(); }); + // --- messages --- // + it('should emit parsed messages', () => { const received: any[] = []; client.subscribe((m) => received.push(m)); @@ -158,4 +253,12 @@ describe('LightningWebSocketClient', () => { close(); jest.advanceTimersByTime(RETRY_WAIT_MS); } + + // Hold an open connection for a while, answering the heartbeat as a live peer would. + function hold(durationMs: number) { + for (let left = durationMs; left > 0; left -= PING_INTERVAL_MS) { + jest.advanceTimersByTime(Math.min(left, PING_INTERVAL_MS)); + latest().emit('pong'); + } + } }); diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index d9155d3a8c..1c661c8a2e 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -10,10 +10,17 @@ export class LightningWebSocketClient extends Subject { private webSocket: WebSocket; private readonly retryWaitTimeSec = 10; + private readonly handshakeTimeoutSec = 15; + private readonly pingIntervalSec = 30; private readonly alertWindowSec = 3600; private readonly retriesPerAlert = 30; + private readonly maxDisconnectedSec = 300; private retryTimestamps: number[] = []; + private disconnectedSince?: number; + private retryFailureLogged = false; + private pingTimer?: NodeJS.Timeout; + private awaitingPong = false; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -32,6 +39,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, }, @@ -46,6 +57,10 @@ export class LightningWebSocketClient extends Subject { socket.on('open', () => { this.logger.info(`WebSocket ${this.wsUrl}: open`); + this.disconnectedSince = undefined; + this.retryFailureLogged = false; + this.startHeartbeat(socket); + socket.send(JSON.stringify(openRequestBody)); }); @@ -56,6 +71,7 @@ export class LightningWebSocketClient extends Subject { socket.on('close', () => { this.logger.info(`WebSocket ${this.wsUrl}: close`); + this.stopHeartbeat(); this.scheduleRetry(openRequestBody); }); @@ -70,25 +86,62 @@ export class LightningWebSocketClient extends Subject { socket.on('ping', (pingMessage: any) => { socket.pong(pingMessage); }); + + socket.on('pong', () => { + this.awaitingPong = false; + }); + } + + // A half-open connection produces no traffic and no 'close', so it would sit there delivering + // nothing while every signal below reads it as a healthy, quiet stream. Pinging ourselves rather + // than watching for inbound traffic keeps this independent of how often LND has something to say. + private startHeartbeat(socket: WebSocket) { + this.stopHeartbeat(); + this.awaitingPong = false; + + this.pingTimer = setInterval(() => { + if (this.awaitingPong) { + this.logger.error(`WebSocket ${this.wsUrl}: no pong within ${this.pingIntervalSec} sec., reconnecting`); + + return socket.terminate(); + } + + this.awaitingPong = true; + socket.ping(); + }, this.pingIntervalSec * 1000); + } + + private stopHeartbeat() { + if (this.pingTimer) clearInterval(this.pingTimer); + this.pingTimer = undefined; } // 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. // - // How often a stream reconnects is what separates a healthy one — LND restarts, deploys — from - // one that cannot hold a connection, and unlike a run of consecutive failures it does not reset - // itself on the occasional successful cycle. LND sends no backlog on subscribe, so every gap - // drops events for good; a stream reconnecting this often is losing data either way. + // Two signals, because neither covers the other. How long the stream has been down catches one + // that cannot connect at all, however slowly each attempt fails — a connect that times out after + // two minutes reconnects too rarely to register as a rate. How often it reconnects catches one + // that connects and drops straight back, which keeps the downtime short but delivers nothing: + // LND sends no backlog on subscribe, so every gap loses events for good. private scheduleRetry(openRequestBody: any) { - const now = Date.now(); + // Monotonic: an NTP or DST correction must not fabricate or discard a window's worth of history. + const now = performance.now(); + + this.disconnectedSince ??= now; + if (now - this.disconnectedSince >= this.maxDisconnectedSec * 1000) { + this.logger.error(`WebSocket ${this.wsUrl}: no connection for ${this.maxDisconnectedSec} sec.`); + + this.disconnectedSince = now; + } + this.retryTimestamps = this.retryTimestamps.filter((t) => now - t < this.alertWindowSec * 1000); this.retryTimestamps.push(now); if (this.retryTimestamps.length >= this.retriesPerAlert) { - this.logger.error( - `WebSocket ${this.wsUrl}: ${this.retryTimestamps.length} reconnects within ${this.alertWindowSec} sec.`, - ); + const spanSec = Math.round((now - this.retryTimestamps[0]) / 1000); + this.logger.error(`WebSocket ${this.wsUrl}: ${this.retryTimestamps.length} reconnects within ${spanSec} sec.`); this.retryTimestamps = []; } @@ -102,8 +155,12 @@ export class LightningWebSocketClient extends Subject { } 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. - this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); + // no socket created no 'close' event would ever drive the next attempt. Log the cause once + // and let the signals above carry the repetition. + if (!this.retryFailureLogged) { + this.retryFailureLogged = true; + this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); + } this.scheduleRetry(openRequestBody); } From ae18576c8ebc3ea5c842cb7dadbb7866308f059c Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 14:27:19 -0300 Subject: [PATCH 06/10] fix(lightning): drop the ping heartbeat, throttle the connect error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The heartbeat added in the previous commit rests on LND answering client pings, and that does not hold. lnd's WebSocket proxy installs no ping handler, so a ping is answered only by gorilla's default, which runs while the read loop is active and writes the pong with a one second deadline — and a write timeout is treated as temporary and swallowed. Tolerating exactly one missed pong then makes any event-loop stall longer than the interval enough to terminate a healthy stream, which loses events because LND sends no backlog on subscribe. Detecting a half-open connection is worth doing, but not on an assumption that cannot be checked from here and whose failure mode is a reconnect storm on all three streams at once. Left as a follow-up, noted in the PR. handshakeTimeout stays: a stalled upgrade emitting neither 'error' nor 'close' is the one liveness gap that needed no assumption about the peer. Retrying forever also made the socket 'error' handler unbounded. Previously the client gave up after 30 attempts and went quiet; now an unreachable LND logs at error level every 10 sec. indefinitely, which is the same defect the last commit fixed one line away for the construction failure. Both are now reported once per outage, and the throttled signals carry the repetition. Also: - report the downtime actually observed rather than the configured threshold, which was constant by construction — the same fix the last commit applied to the reconnect message but not to its sibling. Throttling now uses its own field, so re-arming no longer discards when the outage started - clear the construction latch on a successful construction rather than on a later 'open': construction can start succeeding while LND is still down, so 'open' never arrives and a second, different failure went unreported Every branch is now mutation-checked: negating either signal, dropping either throttle, either re-arm, the handshake timeout, the monotonic clock or the socket binding each break at least one test. 21 tests fail against develop, 5 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 137 ++++++++++-------- .../lightning/lightning-ws-client.ts | 64 +++----- 2 files changed, 102 insertions(+), 99 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index 09d41bdccd..d822bd449e 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -14,9 +14,7 @@ jest.mock('ws', () => ({ const socket = new EventEmitter(); socket.send = jest.fn(); - socket.ping = jest.fn(); socket.pong = jest.fn(); - socket.terminate = jest.fn(() => socket.emit('close')); mockSockets.push(socket); return socket; @@ -26,7 +24,6 @@ jest.mock('ws', () => ({ describe('LightningWebSocketClient', () => { const RETRIES_PER_ALERT = 30; const RETRY_WAIT_MS = 10000; - const PING_INTERVAL_MS = 30000; const MAX_DISCONNECTED_MS = 300000; const ALERT_WINDOW_MS = 3600000; const URL = 'wss://test/v1/invoices/subscribe'; @@ -78,30 +75,48 @@ describe('LightningWebSocketClient', () => { // --- signal: how long the stream has been down --- // - // Immune to how fast each attempt fails. A connect that times out after two minutes reconnects - // far too rarely to register as a rate, and such a stream is completely dead. + // 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', () => { - const slowFailureMs = 130000; - for (let i = 0; i < 10; i++) { - jest.advanceTimersByTime(slowFailureMs); + jest.advanceTimersByTime(130000); reconnect(); } - expect(logger.error).toHaveBeenCalledWith(`WebSocket ${URL}: no connection for 300 sec.`); + 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); - jest.advanceTimersByTime(RETRY_WAIT_MS); + jest.advanceTimersByTime(MAX_DISCONNECTED_MS + RETRY_WAIT_MS); open(); logger.error.mockClear(); close(); jest.advanceTimersByTime(RETRY_WAIT_MS); - expect(logger.error).not.toHaveBeenCalledWith(expect.stringContaining('no connection for')); + expect(errorsMatching(/no connection for/)).toHaveLength(0); }); // --- signal: how often the stream reconnects --- // @@ -134,49 +149,18 @@ describe('LightningWebSocketClient', () => { it('should report reconnect bursts at most once per threshold', () => { for (let i = 0; i < RETRIES_PER_ALERT * 2; i++) reconnect(); - expect(logger.error.mock.calls.filter((c) => /reconnects within/.test(c[0]))).toHaveLength(2); - }); - - // --- liveness --- // - - // A half-open connection yields no traffic and no 'close', so both signals above would read it as - // a healthy quiet stream forever. - it('should reconnect when pings go unanswered', () => { - open(); - const stalled = latest(); - - jest.advanceTimersByTime(PING_INTERVAL_MS); - expect(stalled.ping).toHaveBeenCalledTimes(1); - - jest.advanceTimersByTime(PING_INTERVAL_MS); - expect(stalled.terminate).toHaveBeenCalled(); - expect(logger.error).toHaveBeenCalledWith(expect.stringContaining('no pong')); - - jest.advanceTimersByTime(RETRY_WAIT_MS); - expect(mockSockets).toHaveLength(2); + expect(errorsMatching(/reconnects within/)).toHaveLength(2); }); - it('should keep a connection that answers pings', () => { - open(); - const socket = latest(); - - for (let i = 0; i < 10; i++) { - jest.advanceTimersByTime(PING_INTERVAL_MS); - socket.emit('pong'); - } + // 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(); - expect(socket.terminate).not.toHaveBeenCalled(); - expect(mockSockets).toHaveLength(1); - }); - - it('should stop pinging a closed connection', () => { - open(); - const socket = latest(); - close(); - - jest.advanceTimersByTime(PING_INTERVAL_MS * 3); + hold(ALERT_WINDOW_MS * 2); + jest.setSystemTime(new Date(Date.now() - 3 * 3600000)); + reconnect(); - expect(socket.ping).not.toHaveBeenCalled(); + expect(errorsMatching(/reconnects within/)).toHaveLength(0); }); // --- retry robustness --- // @@ -203,7 +187,44 @@ describe('LightningWebSocketClient', () => { close(); jest.advanceTimersByTime(RETRY_WAIT_MS * 10); - expect(logger.error.mock.calls.filter((c) => /retry failed/.test(c[0]))).toHaveLength(1); + 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 again'); + close(); + jest.advanceTimersByTime(RETRY_WAIT_MS * 3); + + expect(errorsMatching(/retry failed/)).toHaveLength(2); + }); + + // 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); }); it('should not let a retired socket drive its successor', () => { @@ -254,11 +275,13 @@ describe('LightningWebSocketClient', () => { jest.advanceTimersByTime(RETRY_WAIT_MS); } - // Hold an open connection for a while, answering the heartbeat as a live peer would. function hold(durationMs: number) { - for (let left = durationMs; left > 0; left -= PING_INTERVAL_MS) { - jest.advanceTimersByTime(Math.min(left, PING_INTERVAL_MS)); - latest().emit('pong'); - } + jest.advanceTimersByTime(durationMs); + } + + 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 1c661c8a2e..e7bb1fb6cd 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -11,16 +11,15 @@ export class LightningWebSocketClient extends Subject { private readonly retryWaitTimeSec = 10; private readonly handshakeTimeoutSec = 15; - private readonly pingIntervalSec = 30; private readonly alertWindowSec = 3600; private readonly retriesPerAlert = 30; private readonly maxDisconnectedSec = 300; private retryTimestamps: number[] = []; private disconnectedSince?: number; - private retryFailureLogged = false; - private pingTimer?: NodeJS.Timeout; - private awaitingPong = false; + private lastDowntimeAlert?: number; + private connectErrorLogged = false; + private constructionErrorLogged = false; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -58,20 +57,24 @@ export class LightningWebSocketClient extends Subject { this.logger.info(`WebSocket ${this.wsUrl}: open`); this.disconnectedSince = undefined; - this.retryFailureLogged = false; - this.startHeartbeat(socket); + this.lastDowntimeAlert = undefined; + this.connectErrorLogged = false; socket.send(JSON.stringify(openRequestBody)); }); socket.on('error', (err: any) => { + // Retrying forever means this fires every retryWaitTimeSec for as long as LND is unreachable. + // Report the cause once per outage and let the throttled signals below carry the repetition. + if (this.connectErrorLogged) return; + + this.connectErrorLogged = true; this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); socket.on('close', () => { this.logger.info(`WebSocket ${this.wsUrl}: close`); - this.stopHeartbeat(); this.scheduleRetry(openRequestBody); }); @@ -86,34 +89,6 @@ export class LightningWebSocketClient extends Subject { socket.on('ping', (pingMessage: any) => { socket.pong(pingMessage); }); - - socket.on('pong', () => { - this.awaitingPong = false; - }); - } - - // A half-open connection produces no traffic and no 'close', so it would sit there delivering - // nothing while every signal below reads it as a healthy, quiet stream. Pinging ourselves rather - // than watching for inbound traffic keeps this independent of how often LND has something to say. - private startHeartbeat(socket: WebSocket) { - this.stopHeartbeat(); - this.awaitingPong = false; - - this.pingTimer = setInterval(() => { - if (this.awaitingPong) { - this.logger.error(`WebSocket ${this.wsUrl}: no pong within ${this.pingIntervalSec} sec., reconnecting`); - - return socket.terminate(); - } - - this.awaitingPong = true; - socket.ping(); - }, this.pingIntervalSec * 1000); - } - - private stopHeartbeat() { - if (this.pingTimer) clearInterval(this.pingTimer); - this.pingTimer = undefined; } // Retry forever. The process used to die whenever a stream broke and came back with fresh @@ -130,10 +105,13 @@ export class LightningWebSocketClient extends Subject { const now = performance.now(); this.disconnectedSince ??= now; - if (now - this.disconnectedSince >= this.maxDisconnectedSec * 1000) { - this.logger.error(`WebSocket ${this.wsUrl}: no connection for ${this.maxDisconnectedSec} sec.`); + 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.disconnectedSince = now; + this.lastDowntimeAlert = now; } this.retryTimestamps = this.retryTimestamps.filter((t) => now - t < this.alertWindowSec * 1000); @@ -152,13 +130,15 @@ export class LightningWebSocketClient extends Subject { try { this.createWebSocket(); this.setup(openRequestBody); + + this.constructionErrorLogged = false; } 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. Log the cause once - // and let the signals above carry the repetition. - if (!this.retryFailureLogged) { - this.retryFailureLogged = true; + // no socket created no 'close' event would ever drive the next attempt. Latch on the + // construction rather than on a later 'open', so a second, different failure still reports. + if (!this.constructionErrorLogged) { + this.constructionErrorLogged = true; this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); } From ca58f3443e9ee0c9c5cbb13d3316192edbf29409 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 14:46:51 -0300 Subject: [PATCH 07/10] fix(lightning): report a connect error whose cause changes mid-outage MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `connectErrorLogged` was a boolean cleared only on 'open', so during an outage that never opens, only the first cause was ever recorded. LND going down logs ECONNREFUSED; if it comes back with a rotated macaroon, every upgrade is rejected with a 401 that never reaches the log, and the operator keeps seeing an hour-old network error and chases a fault that no longer exists. That is the same asymmetry the previous commit fixed one line away for the construction failure. Both latches now key on the cause rather than on a boolean, so a repeat stays quiet and a change is reported. The set of causes is small and bounded — refused, timed out, unreachable, handshake rejected — so this cannot grow into per-attempt logging. Also closes three gaps that mutation testing found unpinned: - resetting the outage start on 'open'. Without it a fresh five-minute outage after a day of uptime reports the time since the process started, which is the constant-by-construction fault this branch has now fixed twice elsewhere - sending the subscribe body on 'open'. Dropping it leaves the client connected and silent, LND with nothing to send back, and both health signals reading a dead stream as healthy and quiet — the most complete silent death in the file - sending it on the socket that opened rather than on the field. Only the pong half of that binding was covered, though both are load-bearing: ws throws synchronously from send() and pong() while a socket is still connecting Also drops a test helper left over from the removed heartbeat, and a comment citing a two-minute connect that handshakeTimeout now bounds. 23 tests fail against develop, 2 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 56 ++++++++++++++++--- .../lightning/lightning-ws-client.ts | 35 +++++++----- 2 files changed, 70 insertions(+), 21 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index d822bd449e..3bbe97c6f5 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -119,6 +119,18 @@ describe('LightningWebSocketClient', () => { 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 @@ -126,7 +138,7 @@ describe('LightningWebSocketClient', () => { it('should report a stream that reconnects constantly despite good cycles', () => { for (let i = 0; i < RETRIES_PER_ALERT; i++) { open(); - hold(60000); + jest.advanceTimersByTime(60000); reconnect(); } @@ -138,7 +150,7 @@ describe('LightningWebSocketClient', () => { it('should not report a stream that reconnects rarely', () => { for (let i = 0; i < RETRIES_PER_ALERT + 5; i++) { open(); - hold(ALERT_WINDOW_MS); + jest.advanceTimersByTime(ALERT_WINDOW_MS); reconnect(); } @@ -156,7 +168,7 @@ describe('LightningWebSocketClient', () => { it('should measure the reconnect window on a monotonic clock', () => { for (let i = 0; i < RETRIES_PER_ALERT - 1; i++) reconnect(); - hold(ALERT_WINDOW_MS * 2); + jest.advanceTimersByTime(ALERT_WINDOW_MS * 2); jest.setSystemTime(new Date(Date.now() - 3 * 3600000)); reconnect(); @@ -200,13 +212,24 @@ describe('LightningWebSocketClient', () => { mockConstructorError = undefined; jest.advanceTimersByTime(RETRY_WAIT_MS); - mockConstructorError = new Error('cert unavailable again'); + 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); + }); + // 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++) { @@ -227,14 +250,35 @@ describe('LightningWebSocketClient', () => { expect(errorsMatching(/: error$/)).toHaveLength(2); }); + // 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 --- // @@ -275,10 +319,6 @@ describe('LightningWebSocketClient', () => { jest.advanceTimersByTime(RETRY_WAIT_MS); } - function hold(durationMs: number) { - jest.advanceTimersByTime(durationMs); - } - function errorsMatching(pattern: RegExp): RegExpMatchArray[] { return logger.error.mock.calls .map((c) => String(c[0]).match(pattern)) diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index e7bb1fb6cd..7dbe76a62e 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -4,6 +4,10 @@ import { GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; import WebSocket from 'ws'; +function errorCause(e: any): string { + return e?.message ?? String(e); +} + export class LightningWebSocketClient extends Subject { private readonly logger = new LightningLogger(LightningWebSocketClient); @@ -18,8 +22,8 @@ export class LightningWebSocketClient extends Subject { private retryTimestamps: number[] = []; private disconnectedSince?: number; private lastDowntimeAlert?: number; - private connectErrorLogged = false; - private constructionErrorLogged = false; + private lastConnectError?: string; + private lastConstructionError?: string; constructor(private wsUrl: string, private macaroon: string) { super(); @@ -58,17 +62,20 @@ export class LightningWebSocketClient extends Subject { this.disconnectedSince = undefined; this.lastDowntimeAlert = undefined; - this.connectErrorLogged = false; + this.lastConnectError = undefined; socket.send(JSON.stringify(openRequestBody)); }); socket.on('error', (err: any) => { // Retrying forever means this fires every retryWaitTimeSec for as long as LND is unreachable. - // Report the cause once per outage and let the throttled signals below carry the repetition. - if (this.connectErrorLogged) return; + // Report each distinct cause once and let the throttled signals below carry the repetition — + // latching on a boolean would hide a cause that changes mid-outage, such as a refused + // connection becoming a rejected macaroon, which never produces the 'open' that clears it. + const cause = errorCause(err); + if (this.lastConnectError === cause) return; - this.connectErrorLogged = true; + this.lastConnectError = cause; this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); @@ -96,8 +103,8 @@ export class LightningWebSocketClient extends Subject { // lifetime of the process, with nothing erroring and nothing alerting. // // Two signals, because neither covers the other. How long the stream has been down catches one - // that cannot connect at all, however slowly each attempt fails — a connect that times out after - // two minutes reconnects too rarely to register as a rate. How often it reconnects catches one + // that cannot connect at all, however slowly each attempt fails, which can be far too rarely to + // register as a rate. How often it reconnects catches one // that connects and drops straight back, which keeps the downtime short but delivers nothing: // LND sends no backlog on subscribe, so every gap loses events for good. private scheduleRetry(openRequestBody: any) { @@ -131,14 +138,16 @@ export class LightningWebSocketClient extends Subject { this.createWebSocket(); this.setup(openRequestBody); - this.constructionErrorLogged = false; + this.lastConstructionError = undefined; } 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. Latch on the - // construction rather than on a later 'open', so a second, different failure still reports. - if (!this.constructionErrorLogged) { - this.constructionErrorLogged = true; + // 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. + const cause = errorCause(e); + if (this.lastConstructionError !== cause) { + this.lastConstructionError = cause; this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); } From 4cbc044a912b215e90aac827376292ccb715ded7 Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 15:08:09 -0300 Subject: [PATCH 08/10] fix(lightning): bound the error log by cause set, not by the last cause MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous commit keyed both error latches on the cause so that a change mid outage still reported, and justified it with "the set of causes is small and bounded, so this cannot grow into per-attempt logging". That does not follow: a bounded set of size two, alternating, defeats a single remembered value entirely. A crash-looping LND produces exactly that on the retry cadence — refused while the process is down, rejected while the REST gateway boots — and measured on the branch it logged on all 100 of 100 attempts, restoring the per-attempt volume that commit set out to remove. Each outage now remembers the causes it has already reported, capped, so a repeat stays quiet, a change is still reported, and neither alternation nor a message that varies can push it back to one line per attempt. errorCause also needed hardening, since it runs inside the catch that keeps the reconnect chain alive: - it could throw. String() throws on a null-prototype object, and a throw there skips the reschedule and escapes the timer, which is a dead stream plus an uncaught exception — the two failure modes this branch exists to remove. The sibling helper in the service was already guarded against exactly this - it collapsed every connect errno onto one key. From Node 20 autoSelectFamily is on by default and a connect failure to a hostname arrives as an AggregateError whose message is empty, so refused and unreachable were indistinguishable. The code is now part of the key. Latent on the pinned Node 18 base image, live on the next bump Also states the real reason both health signals exist. handshakeTimeout bounds a cycle at 25 sec., so a stream that cannot connect does now register as a rate — the downtime signal earns its place by reporting the size of an outage, reaching one twice as fast, and not depending on that bound holding. 26 tests fail against develop, 5 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 55 +++++++++++++++++++ .../lightning/lightning-ws-client.ts | 53 +++++++++++------- 2 files changed, 89 insertions(+), 19 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index 3bbe97c6f5..b973e463d7 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -25,6 +25,7 @@ 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'; @@ -230,6 +231,17 @@ describe('LightningWebSocketClient', () => { 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/).length).toBeLessThanOrEqual(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++) { @@ -250,6 +262,49 @@ describe('LightningWebSocketClient', () => { 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$/).length).toBeLessThanOrEqual(MAX_REPORTED_CAUSES); + }); + + // 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', () => { diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 7dbe76a62e..e366712e69 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -4,8 +4,17 @@ import { GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; import WebSocket from 'ws'; +// 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 { - return e?.message ?? String(e); + 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 { @@ -19,11 +28,13 @@ export class LightningWebSocketClient extends Subject { private readonly retriesPerAlert = 30; private readonly maxDisconnectedSec = 300; + private readonly maxReportedCauses = 5; + private retryTimestamps: number[] = []; private disconnectedSince?: number; private lastDowntimeAlert?: number; - private lastConnectError?: string; - private lastConstructionError?: string; + private readonly reportedConnectErrors = new Set(); + private readonly reportedConstructionErrors = new Set(); constructor(private wsUrl: string, private macaroon: string) { super(); @@ -62,20 +73,18 @@ export class LightningWebSocketClient extends Subject { this.disconnectedSince = undefined; this.lastDowntimeAlert = undefined; - this.lastConnectError = undefined; + this.reportedConnectErrors.clear(); socket.send(JSON.stringify(openRequestBody)); }); socket.on('error', (err: any) => { // Retrying forever means this fires every retryWaitTimeSec for as long as LND is unreachable. - // Report each distinct cause once and let the throttled signals below carry the repetition — - // latching on a boolean would hide a cause that changes mid-outage, such as a refused - // connection becoming a rejected macaroon, which never produces the 'open' that clears it. - const cause = errorCause(err); - if (this.lastConnectError === cause) return; + // 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.reportedConnectErrors, errorCause(err))) return; - this.lastConnectError = cause; this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); @@ -98,15 +107,23 @@ export class LightningWebSocketClient extends Subject { }); } + private shouldReport(reported: Set, cause: string): boolean { + if (reported.has(cause) || reported.size >= this.maxReportedCauses) return false; + + reported.add(cause); + + return true; + } + // 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, because neither covers the other. How long the stream has been down catches one - // that cannot connect at all, however slowly each attempt fails, which can be far too rarely to - // register as a rate. How often it reconnects catches one - // that connects and drops straight back, which keeps the downtime short but delivers nothing: - // LND sends no backlog on subscribe, so every gap loses events for good. + // 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(); @@ -138,16 +155,14 @@ export class LightningWebSocketClient extends Subject { this.createWebSocket(); this.setup(openRequestBody); - this.lastConstructionError = undefined; + this.reportedConstructionErrors.clear(); } 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. - const cause = errorCause(e); - if (this.lastConstructionError !== cause) { - this.lastConstructionError = cause; + if (this.shouldReport(this.reportedConstructionErrors, errorCause(e))) { this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); } From 03d6eaa828f3a51978c4f7aead4a713c8b8ee4bf Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 15:26:34 -0300 Subject: [PATCH 09/10] fix(lightning): say so when an outage produces more causes than it reports MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Capping the reported causes stopped the per-attempt logging, but saturating is silent: whichever five causes arrive first become the whole story. Measured with five gateway errors followed by an expired macaroon, the 401 never appeared once across 200 attempts, and nothing in the log said anything had been dropped — the operator reads five transient, self-healing-looking causes and concludes LND is restarting. That is the wrong-fault trap the cause set was meant to remove, narrowed from one stale cause to five rather than closed. A saturated set now says so once. Detection was never affected; this is about the log being honest about its own completeness. Also tightens two test assertions that were too loose to pin what they claimed: the cap tests bounded the count from above only, so a cap of 2, 3 or 4 — or a regression silencing the log entirely — passed the whole suite. And errors carrying neither a code nor a message now have a test, since without the String() fallback they would all collapse onto one key and only the first would be logged. Two mutants inside the cause-key helper survive and are left there deliberately: dropping the empty-string filter is behaviourally equivalent, and dropping the message from the key needs two errors that share a code but differ in message, which cannot arise against a fixed URL. 27 tests fail against develop, 1 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 23 +++++++++++++++++-- .../lightning/lightning-ws-client.ts | 17 +++++++++++++- 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index b973e463d7..4c9c255bfc 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -239,7 +239,7 @@ describe('LightningWebSocketClient', () => { jest.advanceTimersByTime(RETRY_WAIT_MS); } - expect(errorsMatching(/retry failed/).length).toBeLessThanOrEqual(MAX_REPORTED_CAUSES); + expect(errorsMatching(/retry failed/)).toHaveLength(MAX_REPORTED_CAUSES); }); // Retrying forever means this fires every 10 sec. for as long as LND is unreachable. @@ -279,7 +279,26 @@ describe('LightningWebSocketClient', () => { reconnect(); } - expect(errorsMatching(/: error$/).length).toBeLessThanOrEqual(MAX_REPORTED_CAUSES); + 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 error causes suppressed/)).toHaveLength(1); + }); + + 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 diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index e366712e69..919debabcc 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -4,6 +4,9 @@ import { GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; import WebSocket from 'ws'; +// Marks a cause set as saturated. A NUL cannot appear in an error message, so this cannot collide. +const SATURATION_KEY = '\0saturated'; + // 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. @@ -108,7 +111,19 @@ export class LightningWebSocketClient extends Subject { } private shouldReport(reported: Set, cause: string): boolean { - if (reported.has(cause) || reported.size >= this.maxReportedCauses) return false; + if (reported.has(cause)) return false; + + if (reported.size >= this.maxReportedCauses) { + // Saturating silently would be worse than the stale single cause this replaced: whichever + // causes happened to come first read as the complete list, and the one that actually needs a + // human — an expired macaroon behind a burst of gateway errors — never appears at all. + if (!reported.has(SATURATION_KEY)) { + reported.add(SATURATION_KEY); + this.logger.error(`WebSocket ${this.wsUrl}: further error causes suppressed this outage`); + } + + return false; + } reported.add(cause); From c2fca3aff27cecdd7d9862ac2073735497b837af Mon Sep 17 00:00:00 2001 From: Daniel Padrino Date: Wed, 29 Jul 2026 15:46:27 -0300 Subject: [PATCH 10/10] fix(lightning): prove the mapping guard on all three streams, name the suppressed log MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The guard this branch exists for was pinned on one stream of three. Removing safeMap from the onchain and payment pipes left the whole service suite green, because those two mappers cannot throw on any input the tests fed them. The frame `null` is legal — `JSON.parse('null')` — and `null.result` throws in all three, which is what tells an unguarded pipe from a guarded one. Without it the headline claim rested on the invoice path alone. The suppression notice could not say what it was about. Both cause sets share the reporting helper and emitted identical text, so two saturating in one outage read as a doubled line rather than one truncated connect log and one truncated retry log. It now names which. Dropped "this outage" as well: the construction set is cleared by a successful construction, not by the end of an outage, so the phrase was wrong on that path. Saturation is now tracked beside the causes instead of as an entry among them. The sentinel's justifying comment was wrong — a JavaScript string holds a NUL perfectly well, and an error message that happened to equal the marker would have silently disabled the notice, which is the failure it was added to remove. Two claims of mine in the previous commit did not survive review, both corrected here rather than argued: - "dropping the message from the cause key needs two errors that share a code but differ in message, which cannot arise against a fixed URL" is false. read and write ECONNRESET share a code and differ in message with no DNS involved, and Node's connect message carries the resolved address, which changes when a container is recreated mid-outage. The mutant is killed by a test now - errorCause was unpinned at the construction call site — the one its own comment is written for — so substituting String() there passed the whole suite while reintroducing the throw that skips the reschedule 32 tests fail against develop, 2 against the preceding commit. --- .../__test__/lightning-ws-client.spec.ts | 45 ++++++++++++++++++- .../lightning/lightning-ws-client.ts | 45 +++++++++++-------- .../__test__/lightning-ws-service.spec.ts | 27 +++++++++++ 3 files changed, 98 insertions(+), 19 deletions(-) diff --git a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts index 4c9c255bfc..1526bf453b 100644 --- a/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts +++ b/src/integration/blockchain/lightning/__test__/lightning-ws-client.spec.ts @@ -290,7 +290,50 @@ describe('LightningWebSocketClient', () => { reconnect(); } - expect(errorsMatching(/further error causes suppressed/)).toHaveLength(1); + 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', () => { diff --git a/src/integration/blockchain/lightning/lightning-ws-client.ts b/src/integration/blockchain/lightning/lightning-ws-client.ts index 919debabcc..0404e67389 100644 --- a/src/integration/blockchain/lightning/lightning-ws-client.ts +++ b/src/integration/blockchain/lightning/lightning-ws-client.ts @@ -4,8 +4,12 @@ import { GetConfig } from 'src/config/config'; import { LightningLogger } from 'src/shared/services/lightning-logger'; import WebSocket from 'ws'; -// Marks a cause set as saturated. A NUL cannot appear in an error message, so this cannot collide. -const SATURATION_KEY = '\0saturated'; +// 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 @@ -36,8 +40,8 @@ export class LightningWebSocketClient extends Subject { private retryTimestamps: number[] = []; private disconnectedSince?: number; private lastDowntimeAlert?: number; - private readonly reportedConnectErrors = new Set(); - private readonly reportedConstructionErrors = new Set(); + 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(); @@ -76,7 +80,7 @@ export class LightningWebSocketClient extends Subject { this.disconnectedSince = undefined; this.lastDowntimeAlert = undefined; - this.reportedConnectErrors.clear(); + this.clearReported(this.connectErrors); socket.send(JSON.stringify(openRequestBody)); }); @@ -86,7 +90,7 @@ export class LightningWebSocketClient extends Subject { // 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.reportedConnectErrors, errorCause(err))) return; + if (!this.shouldReport(this.connectErrors, errorCause(err), 'connection error')) return; this.logger.error(`WebSocket ${this.wsUrl}: error`, err); }); @@ -110,26 +114,31 @@ export class LightningWebSocketClient extends Subject { }); } - private shouldReport(reported: Set, cause: string): boolean { - if (reported.has(cause)) return false; + private shouldReport(reported: ReportedCauses, cause: string, label: string): boolean { + if (reported.causes.has(cause)) return false; - if (reported.size >= this.maxReportedCauses) { - // Saturating silently would be worse than the stale single cause this replaced: whichever - // causes happened to come first read as the complete list, and the one that actually needs a - // human — an expired macaroon behind a burst of gateway errors — never appears at all. - if (!reported.has(SATURATION_KEY)) { - reported.add(SATURATION_KEY); - this.logger.error(`WebSocket ${this.wsUrl}: further error causes suppressed this outage`); + 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.add(cause); + 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. @@ -170,14 +179,14 @@ export class LightningWebSocketClient extends Subject { this.createWebSocket(); this.setup(openRequestBody); - this.reportedConstructionErrors.clear(); + 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.reportedConstructionErrors, errorCause(e))) { + if (this.shouldReport(this.constructionErrors, errorCause(e), 'retry failure')) { this.logger.error(`WebSocket ${this.wsUrl}: retry failed`, e); } 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 index 34347c00d7..92c6c07b68 100644 --- a/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts +++ b/src/integration/blockchain/lightning/services/__test__/lightning-ws-service.spec.ts @@ -72,6 +72,33 @@ describe('LightningWebSocketService', () => { 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);