fix(streams): harden wasm elements per review

Five open review-round findings on the connection elements:

- Guard per-line/per-frame debug interpolation behind debug.enabled in
  wasm-n2k's decode-error handler and j1939-can's frame handler.
- Protect j1939-can's start()-failure teardown: connect() also runs
  from the reconnect timer, where a throw from stop() would kill the
  process and skip the reconnect.
- wasm-n2k-bytes: register txHandler before connect() so the
  constructor has no initialization-order dependency.
- wasm-n2k-bytes: honor noDataReceivedTimeout on the TCP transport
  with a receive-only idle timer — socket.setTimeout() counts the
  periodic keepalive writes as activity, so it could never fire while
  the gateway sends nothing. Armed per connection, re-armed only on
  received data, cleared in retry and end().
- Maretron help text no longer claims canboatjs handles the framing;
  the wasm variant shares the block.

Author: dirkwa

packages/server-admin-ui/src/views/ServerConfig/BasicProvider.tsx

@@ -1928,7 +1928,7 @@ function NMEA2000({
           </Form.Group>
           <div className="text-muted small mt-1 mb-2">
             Maretron IPG 100 Ethernet gateway. Default TCP port 6543. Uses
-            Maretron&apos;s 0xA5-framed binary protocol (handled by canboatjs).
+            Maretron&apos;s 0xA5-framed binary protocol.
           </div>
         </div>
       )}

packages/streams/src/j1939-can.ts

@@ -107,8 +107,10 @@ export default class J1939Can extends Transform {
           data: msg.data
         })
       } catch (err: unknown) {
-        const message = err instanceof Error ? err.message : String(err)
-        this.debug(`[frame] ${message}`)
+        if (this.debug.enabled) {
+          const message = err instanceof Error ? err.message : String(err)
+          this.debug(`[frame] ${message}`)
+        }
       }
     })
     channel.addListener('onStopped', () => {
@@ -130,8 +132,19 @@ export default class J1939Can extends Transform {
       channel.start()
     } catch (err: unknown) {
       const message = err instanceof Error ? err.message : String(err)
-      channel.removeAllListeners()
-      channel.stop()
+      // connect() also runs from the reconnect timer, where an escaping
+      // throw from the half-started channel's teardown would take the
+      // process down and skip the reconnect.
+      try {
+        channel.removeAllListeners()
+        channel.stop()
+      } catch (stopErr: unknown) {
+        if (this.debug.enabled) {
+          this.debug(
+            `[stop] ${stopErr instanceof Error ? stopErr.message : String(stopErr)}`
+          )
+        }
+      }
       this.channel = null
       if (this.options.providerId) {
         this.options.app.setProviderError?.(this.options.providerId, message)

packages/streams/src/wasm-n2k-bytes.ts

@@ -31,6 +31,7 @@ interface WasmN2kBytesOptions {
   /** Serial transport instead of TCP (e.g. an NGT-1 dongle). */
   device?: string
   baudrate?: number
+  noDataReceivedTimeout?: string | number
   byteKind: 'ngt1' | 'maretron-ipg'
   password?: string
   providerId?: string
@@ -71,6 +72,7 @@ function requireWasm(): WasmByteApi {
 
 const RECONNECT_DELAY = 3000
 const KEEPALIVE_INTERVAL = 20000
+const DEFAULT_IDLE_TIMEOUT_SECONDS = 60
 
 interface ByteTransport {
   write(data: Buffer): void
@@ -84,6 +86,8 @@ export default class WasmN2kBytes extends Transform {
   private socket: ByteTransport | null = null
   private keepaliveTimer: NodeJS.Timeout | null = null
   private reconnectTimer: NodeJS.Timeout | null = null
+  private idleTimer: NodeJS.Timeout | null = null
+  private idleMs = 0
   private stopped = false
   private readonly txHandler: (pgn: unknown) => void
   private readonly debug: DebugLogger
@@ -95,8 +99,6 @@ export default class WasmN2kBytes extends Transform {
     const createDebug = options.createDebug ?? require('debug')
     this.debug = createDebug('signalk:streams:wasm-n2k-bytes')
 
-    this.connect()
-
     this.txHandler = (pgn: unknown) => {
       try {
         const bytes = this.decoder.encodeFrame(JSON.stringify(pgn), true)
@@ -109,6 +111,8 @@ export default class WasmN2kBytes extends Transform {
     }
     options.app.on('nmea2000JsonOut', this.txHandler)
     options.app.emit('nmea2000OutAvailable')
+
+    this.connect()
   }
 
   private status(msg: string): void {
@@ -118,6 +122,31 @@ export default class WasmN2kBytes extends Transform {
     this.debug(msg)
   }
 
+  private clearIdleTimer(): void {
+    if (this.idleTimer) {
+      clearTimeout(this.idleTimer)
+      this.idleTimer = null
+    }
+  }
+
+  private armIdleTimer(): void {
+    this.clearIdleTimer()
+    if (this.idleMs <= 0) {
+      return
+    }
+    const socket = this.socket
+    this.idleTimer = setTimeout(() => {
+      this.idleTimer = null
+      if (!socket || this.socket !== socket) {
+        return
+      }
+      this.debug.enabled && this.debug('Idle timeout, closing socket')
+      // destroy(), not end(): the peer is presumed gone, and only
+      // destroy fires 'close' immediately for the retry path.
+      socket.destroy()
+    }, this.idleMs)
+  }
+
   private connect(): void {
     if (this.stopped) {
       return
@@ -168,12 +197,26 @@ export default class WasmN2kBytes extends Transform {
       }
       const tcp = new Socket()
       this.status(`Connecting to ${host}:${port}`)
+      // Receive-only idle timer: a half-open connection raises neither
+      // 'error' nor 'close', so without it the retry path never runs.
+      // socket.setTimeout() cannot serve here — it counts the periodic
+      // keepalive writes as activity, so it would never fire while the
+      // gateway sends nothing. Re-armed only in the data handler,
+      // cleared in retry and end().
+      const parsedTimeout = Number.parseInt(
+        (this.options.noDataReceivedTimeout + '').trim()
+      )
+      this.idleMs =
+        (isNaN(parsedTimeout) ? DEFAULT_IDLE_TIMEOUT_SECONDS : parsedTimeout) *
+        1000
       tcp.connect(port, host, () => onOpen(tcp))
       socket = tcp
     }
     this.socket = socket
+    this.armIdleTimer()
 
     socket.on('data', (buf: Buffer) => {
+      this.armIdleTimer()
       // Everything here runs inside a socket event handler, where an
       // uncaught throw takes the process down. Malformed framing from
       // a gateway must degrade to a logged error, not a crash.
@@ -206,6 +249,7 @@ export default class WasmN2kBytes extends Transform {
     })
 
     const retry = (why: string) => {
+      this.clearIdleTimer()
       if (this.keepaliveTimer) {
         clearInterval(this.keepaliveTimer)
         this.keepaliveTimer = null
@@ -252,6 +296,7 @@ export default class WasmN2kBytes extends Transform {
       clearInterval(this.keepaliveTimer)
       this.keepaliveTimer = null
     }
+    this.clearIdleTimer()
     this.socket?.destroy()
     this.socket = null
     super.end()

packages/streams/src/wasm-n2k.ts

@@ -107,8 +107,10 @@ export default class WasmN2k extends Transform {
       options.j1939 === true ? { j1939: true } : undefined
     )
     this.fromPgn.on('error', (line: unknown, err: unknown) => {
-      const message = err instanceof Error ? err.message : String(err)
-      this.debug(`[error] ${String(line)} ${message}`)
+      if (this.debug.enabled) {
+        const message = err instanceof Error ? err.message : String(err)
+        this.debug(`[error] ${String(line)} ${message}`)
+      }
       options.app.emit('canboatjs:error', err)
     })