Separate Chaining Keys

README.md

@@ -34,6 +34,8 @@ async function connect() {
     remoteNodePublicKey: '02df5ffe895c778e10f7742a6c5b8a0cefbe9465df58b92fadeb883752c8107c8f',
     // Optional WebSocket proxy endpoint to connect through (see WebSocket Proxy section)
     wsProxy: 'wss://<WEBSOCKET_PROXY>',
+    // Optional TCP Socket to connect through (either use wsProxy OR tcpSocket)
+    tcpSocket: new net.Socket(),
     // The IP address of the node
     ip: '35.232.170.67',
     // The port of the node, defaults to 9735

src/index.ts

@@ -27,6 +27,7 @@ import {
   ConnectionStatus
 } from './types.js'
 
+const MAX_BUFFER_SIZE = 1024 * 1024 * 0.8 // 80% of 1 MB limit imposed by cln commando
 const DEFAULT_RECONNECT_ATTEMPTS = 5
 
 class LnMessage {
@@ -89,6 +90,8 @@ class LnMessage {
   private _messageBuffer: BufferReader
   private _processingBuffer: boolean
   private _l: number | null
+  private _bytesRead: number | null
+  private _bytesWrittenMap: Map<any, number>
 
   constructor(options: LnWebSocketOptions) {
     validateInit(options)
@@ -122,6 +125,8 @@ class LnMessage {
     this.Buffer = Buffer
     this.tcpSocket = tcpSocket
 
+    this._bytesRead = 0
+    this._bytesWrittenMap = new Map()
     this._handshakeState = HANDSHAKE_STATE.INITIATOR_INITIATING
     this._decryptedMsgs$ = new Subject()
     this.decryptedMsgs$ = this._decryptedMsgs$.asObservable()
@@ -177,8 +182,12 @@ class LnMessage {
     }
 
     this.socket.onopen = async () => {
-      this._log('info', 'WebSocket is connected')
+      this._log('info', 'WebSocket is connected at ' + new Date().toISOString())
       this._log('info', 'Creating Act1 message')
+
+      // Resetting bytes read and written counter
+      this._bytesRead = 0
+      this._bytesWrittenMap.set(this.socket, 0)
 
       const msg = this.noise.initiatorAct1(Buffer.from(this.remoteNodePublicKey, 'hex'))
 
@@ -190,7 +199,25 @@ class LnMessage {
     }
 
     this.socket.onclose = async () => {
-      this._log('error', 'WebSocket is closed')
+      this.reconnectWebSocket.bind(this)
+    }
+
+    this.socket.onerror = (err: { message: string }) => {
+      this._log('error', `WebSocket error: ${JSON.stringify(err)}`)
+    }
+
+    this.socket.onmessage = this.queueMessage.bind(this)
+
+    return firstValueFrom(
+      this.connectionStatus$.pipe(
+        filter((status) => status === 'connected' || status === 'disconnected'),
+        map((status) => status === 'connected')
+      )
+    )
+  }
+
+  async reconnectWebSocket () {
+      this._log('error', 'WebSocket is closed at ' + new Date().toISOString())
 
       this.connectionStatus$.next('disconnected')
       this.connected$.next(false)
@@ -207,20 +234,6 @@ class LnMessage {
         this.connect()
         this._attemptedReconnects += 1
       }
-    }
-
-    this.socket.onerror = (err: { message: string }) => {
-      this._log('error', `WebSocket error: ${JSON.stringify(err)}`)
-    }
-
-    this.socket.onmessage = this.queueMessage.bind(this)
-
-    return firstValueFrom(
-      this.connectionStatus$.pipe(
-        filter((status) => status === 'connected' || status === 'disconnected'),
-        map((status) => status === 'connected')
-      )
-    )
   }
 
   private queueMessage(event: { data: ArrayBuffer }) {
@@ -287,9 +300,10 @@ class LnMessage {
         }
       } while (readMore)
     } catch (err) {
-      // Terminate on failures as we won't be able to recovery
+      // Terminate on failures as we won't be able to recover
       // since the noise state has rotated nonce and we won't
       // be able to any more data without additional errors.
+      this._log('error', `Noise state has rotated nonce: ${JSON.stringify(err)}`)
       this.disconnect()
     }
 
@@ -466,6 +480,15 @@ class LnMessage {
 
         case MessageType.CommandoResponse: {
           this._commandoMsgs$.next(payload as CommandoMessage)
+
+          // Counting bytes written and reconnecting if it is about to reach the max limit
+          const currentBytesWritten = this._bytesWrittenMap.get(this.socket) || 0
+          const updatedBytesWritten = currentBytesWritten + Buffer.byteLength(decrypted)
+          this._bytesWrittenMap.set(this.socket, updatedBytesWritten)
+          if (this.connectionStatus$.value === 'connected' && updatedBytesWritten > MAX_BUFFER_SIZE) {
+              this._log('error', 'Bytes received exceeded the limit. Resetting the connection...' + new Date().toISOString())
+              this.reconnectWebSocket()
+          }
         }
 
         // ignore all other messages
@@ -479,10 +502,9 @@ class LnMessage {
     method,
     params = [],
     rune,
-    reqId
+    reqId,
+    reqIdPrefix = 'lnmessage'
   }: CommandoRequest): Promise<JsonRpcSuccessResponse['result']> {
-    this._log('info', `Commando request method: ${method} params: ${JSON.stringify(params)}`)
-
     // not connected, so initiate a connection
     if (this.connectionStatus$.value === 'disconnected') {
       this._log('info', 'No socket connection, so creating one now')
@@ -502,6 +524,8 @@ class LnMessage {
       await firstValueFrom(this.connectionStatus$.pipe(filter((status) => status === 'connected')))
     }
 
+    this._log('info', `Commando request method: ${method} params: ${JSON.stringify(params)}`)
+
     const writer = new BufferWriter()
 
     if (!reqId) {
@@ -516,10 +540,14 @@ class LnMessage {
     // write the id
     writer.writeBytes(Buffer.from(reqId, 'hex'))
 
+    // Unique request id with prefix, method and id
+    const detailedReqId = `${reqIdPrefix}:${method}#${reqId}`
+
     // write the request
     writer.writeBytes(
       Buffer.from(
         JSON.stringify({
+          id: detailedReqId, // Adding id for easier debugging with commando
           rune,
           method,
           params
@@ -534,19 +562,22 @@ class LnMessage {
       this._log('info', 'Sending commando message')
       this.socket.send(message)
 
-      this._log('info', 'Message sent and awaiting response')
+      // Counting bytes read and reconnecting if it is about to reach the max limit
+      this._bytesRead = this._bytesRead ? this._bytesRead + Buffer.byteLength(message) : Buffer.byteLength(message);
+      if (this.connectionStatus$.value === 'connected' && this._bytesRead && this._bytesRead > MAX_BUFFER_SIZE) {
+          this._log('error', 'Bytes sent exceeded the limit. Resetting the connection...' + new Date().toISOString())
+          this.reconnectWebSocket()
+      }
+      this._log('info', `Message sent with id ${detailedReqId} and awaiting response`)
 
       const { response } = await firstValueFrom(
         this._commandoMsgs$.pipe(filter((commandoMsg) => commandoMsg.id === reqId))
       )
 
       const { result } = response as JsonRpcSuccessResponse
       const { error } = response as JsonRpcErrorResponse
-
-      this._log(
-        'info',
-        result ? 'Successful response received' : `Error response received: ${error.message}`
-      )
+
+      this._log('info', result ? `Successful response received for ID: ${response.id}` : `Error response received: ${error.message}`)
 
       if (error) throw error
 
@@ -558,7 +589,7 @@ class LnMessage {
 
   _log(level: keyof Logger, msg: string) {
     if (this._logger && this._logger[level]) {
-      this._logger[level](`[${level.toUpperCase()} - ${new Date().toLocaleTimeString()}]: ${msg}`)
+      this._logger[level](`[${level.toUpperCase()} - ${new Date().toISOString()}]: ${msg}`)
     }
   }
 }

src/types.ts

@@ -179,6 +179,7 @@ export type CommandoRequest = JsonRpcRequest & {
    * request id ahead of time
    */
   reqId?: string
+  reqIdPrefix?: string
 }
 
 export type CommandoResponse = JsonRpcSuccessResponse | JsonRpcErrorResponse