fix(perf): binary WebSocket transport and backpressure fix (#479)

* fix(perf): binary WebSocket transport and backpressure fix (#478)

Fix backpressure triggering on every SSH chunk regardless of buffer
state. The previous implementation used an incorrect heuristic
(eventNames().length * highWaterMark) that always exceeded the
threshold. Replace with actual bufferedAmount reads from the underlying
WebSocket, with high/low water mark hysteresis and drain-based resume.

Emit raw Buffer from SSH2 shell stream instead of converting to UTF-8
string. Socket.IO 4.x sends Buffers as binary WebSocket frames,
eliminating per-chunk toString('utf8') overhead and JSON string
escaping. The exec channel continues to emit strings as it accumulates
output before sending.

Bump webssh2_client to 3.2.2 which handles the binary data event.
Server and client must be updated together — the new server is not
backwards compatible with older clients.

Results: CPU on cat /dev/urandom | hexdump -c dropped from 100% to ~25%.
Normal interactive usage no longer triggers spurious pause/resume.

Closes #478

* refactor(test): extract shared helpers to reduce duplication

Extract setupShellFlow() and createMutableBufferContext() helpers
from repeated test setup blocks in stream-backpressure tests.

Author: billchurch

DOCS/api/WEBSOCKET-API.md

@@ -43,7 +43,10 @@ The server uses Socket.IO for real-time communication. Connect to the WebSocket
 
 4. `data`
    - Emitted when there's output from the SSH session.
-   - Payload: `string` (UTF-8 encoded terminal output)
+   - Payload: `string | Buffer`
+     - **Shell data** is emitted as a raw `Buffer`. Socket.IO 4.x automatically sends Buffers as WebSocket binary frames, avoiding the overhead of UTF-8 string conversion and JSON serialization. The client receives this as an `ArrayBuffer`.
+     - **Exec data** (from the `exec` channel) is emitted as a `string` (UTF-8 encoded), since exec output is already a string from SSH2 and is not a high-throughput path.
+   - Client handling: Convert `ArrayBuffer` to `Uint8Array` and pass directly to `xterm.js` `Terminal.write()`, which natively accepts `Uint8Array`. Only decode to string when text is needed (e.g., session logging).
 
 5. `ssherror`
    - Emitted when an SSH-related error occurs.

DOCS/architecture/CLIENT.md

@@ -75,7 +75,7 @@ The client and server communicate using Socket.IO events:
 
 | Event | Description | Payload |
 |-------|-------------|---------|
-| `data` | Terminal output | `string` |
+| `data` | Terminal output | `string \| ArrayBuffer` (binary for shell, string for exec) |
 | `connect` | Connection established | - |
 | `disconnect` | Connection lost | `{ reason }` |
 | `error` | Error message | `{ message, type }` |
@@ -113,9 +113,14 @@ const socket = io({
   transports: ['websocket', 'polling']
 });
 
-// Event handling
+// Event handling — shell data arrives as binary (ArrayBuffer),
+// exec data arrives as string. xterm.js accepts both Uint8Array and string.
 socket.on('data', (data) => {
-  term.write(data);
+  if (data instanceof ArrayBuffer) {
+    term.write(new Uint8Array(data));
+  } else {
+    term.write(data);
+  }
 });
 
 term.onData((data) => {
@@ -245,7 +250,7 @@ term.loadAddon(searchAddon);
 
 ### WebSocket Optimization
 
-- **Binary frames** for reduced overhead
+- **Binary frames** for shell data — the server emits raw `Buffer` objects for shell output, which Socket.IO sends as binary WebSocket frames. This bypasses UTF-8 string conversion and JSON serialization on the server, and the client passes the resulting `Uint8Array` directly to xterm.js without decoding. String decoding is deferred to the session logger only when logging is active.
 - **Compression** enabled when supported
 - **Reconnection logic** with exponential backoff
 - **Connection pooling** for multiple sessions

app/socket/adapters/service-socket-terminal.ts

@@ -15,6 +15,75 @@ interface TerminalConfig {
   env: Record<string, string>
 }
 
+/** Mutable state for SSH-to-WebSocket backpressure control */
+interface BackpressureState {
+  paused: boolean
+  checkTimerId: ReturnType<typeof setTimeout> | null
+}
+
+const LOW_WATER_MARK_DIVISOR = 4
+
+/**
+ * Safely reads bufferedAmount from the ws WebSocket via the Engine.IO
+ * transport chain. Returns null when unavailable (polling transport,
+ * access failure, or during transport upgrade).
+ */
+export function getWebSocketBufferedBytes(
+  socket: AdapterContext['socket']
+): number | null {
+  try {
+    // Access Engine.IO internals via unknown to avoid coupling to private types
+    // while remaining defensive at runtime
+    const conn: unknown = socket.conn
+    if (typeof conn !== 'object' || conn === null) {
+      return null
+    }
+    const transport: unknown = (conn as Record<string, unknown>)['transport']
+    if (typeof transport !== 'object' || transport === null) {
+      return null
+    }
+    const transportRecord = transport as Record<string, unknown>
+    if (transportRecord['name'] !== 'websocket') {
+      return null
+    }
+    const wsSocket: unknown = transportRecord['socket']
+    if (typeof wsSocket !== 'object' || wsSocket === null) {
+      return null
+    }
+    const amount: unknown = (wsSocket as Record<string, unknown>)['bufferedAmount']
+    if (typeof amount !== 'number') {
+      return null
+    }
+    return amount
+  } catch {
+    return null
+  }
+}
+
+/**
+ * Pure decision function for backpressure control.
+ * Returns 'pause' when buffer exceeds high water mark,
+ * 'resume' when buffer drops below low water mark (HWM / 4),
+ * or 'none' when no action is needed.
+ */
+export function computeBackpressureAction(
+  bufferedBytes: number | null,
+  highWaterMark: number,
+  currentlyPaused: boolean
+): 'pause' | 'resume' | 'none' {
+  if (bufferedBytes === null) {
+    return 'none'
+  }
+  const lowWaterMark = Math.floor(highWaterMark / LOW_WATER_MARK_DIVISOR)
+  if (!currentlyPaused && bufferedBytes >= highWaterMark) {
+    return 'pause'
+  }
+  if (currentlyPaused && bufferedBytes < lowWaterMark) {
+    return 'resume'
+  }
+  return 'none'
+}
+
 export class ServiceSocketTerminal {
   constructor(private readonly context: AdapterContext) {}
 
@@ -180,12 +249,61 @@ export class ServiceSocketTerminal {
 
     // Get rate limit configuration (0 = unlimited)
     const rateLimitBytesPerSec = this.context.config.ssh.outputRateLimitBytesPerSec ?? 0
-    const highWaterMark = this.context.config.ssh.socketHighWaterMark ?? 16384
 
     let rateLimitState:
       | { bytesInWindow: number; windowStart: number; paused: boolean }
       | null = rateLimitBytesPerSec > 0 ? { bytesInWindow: 0, windowStart: Date.now(), paused: false } : null
 
+    // Backpressure state for WebSocket outbound buffer
+    const highWaterMark = this.context.config.ssh.socketHighWaterMark ?? 16384
+    const backpressure: BackpressureState = { paused: false, checkTimerId: null }
+
+    const clearResumeSchedule = (): void => {
+      if (backpressure.checkTimerId !== null) {
+        clearTimeout(backpressure.checkTimerId)
+        backpressure.checkTimerId = null
+      }
+      this.context.socket.conn.removeListener('drain', onDrainCheck)
+    }
+
+    const onDrainCheck = (): void => {
+      checkResume()
+    }
+
+    const checkResume = (): void => {
+      const buffered = getWebSocketBufferedBytes(this.context.socket)
+      const action = computeBackpressureAction(buffered, highWaterMark, backpressure.paused)
+      if (action === 'resume') {
+        backpressure.paused = false
+        clearResumeSchedule()
+        stream.resume()
+        this.context.debug('Backpressure relieved, resuming SSH stream (buffered=%d)', buffered)
+      } else if (backpressure.paused) {
+        // Still above low water mark — re-register for next drain + safety timer
+        scheduleResumeCheck()
+      }
+    }
+
+    const scheduleResumeCheck = (): void => {
+      clearResumeSchedule()
+      this.context.socket.conn.once('drain', onDrainCheck)
+      backpressure.checkTimerId = setTimeout(() => {
+        backpressure.checkTimerId = null
+        checkResume()
+      }, 50)
+    }
+
+    const checkBackpressure = (): void => {
+      const buffered = getWebSocketBufferedBytes(this.context.socket)
+      const action = computeBackpressureAction(buffered, highWaterMark, backpressure.paused)
+      if (action === 'pause') {
+        backpressure.paused = true
+        stream.pause()
+        this.context.debug('Backpressure detected, pausing SSH stream (buffered=%d)', buffered)
+        scheduleResumeCheck()
+      }
+    }
+
     stream.on('data', (chunk: Buffer) => {
       const chunkSize = chunk.length
 
@@ -215,7 +333,9 @@ export class ServiceSocketTerminal {
               rateLimitState.bytesInWindow = 0
               rateLimitState.windowStart = Date.now()
               rateLimitState.paused = false
-              stream.resume()
+              if (!backpressure.paused) {
+                stream.resume()
+              }
               this.context.debug('Rate limit window reset, resuming SSH stream')
             }
           }, delayMs)
@@ -226,23 +346,16 @@ export class ServiceSocketTerminal {
         rateLimitState.bytesInWindow += chunkSize
       }
 
-      // Check Socket.IO buffer backpressure
-      const bufferSize = this.context.socket.eventNames().length * highWaterMark
-      if (bufferSize > highWaterMark * 2) {
-        stream.pause()
-        this.context.debug('Socket.IO buffer high, pausing SSH stream')
+      this.context.socket.emit(SOCKET_EVENTS.SSH_DATA, chunk)
 
-        // Resume when socket drains (use setImmediate to check on next tick)
-        setImmediate(() => {
-          stream.resume()
-          this.context.debug('Socket.IO buffer drained, resuming SSH stream')
-        })
+      // Check backpressure after emit (standard Node.js streams pattern)
+      if (!backpressure.paused) {
+        checkBackpressure()
       }
-
-      this.context.socket.emit(SOCKET_EVENTS.SSH_DATA, chunk.toString('utf8'))
     })
 
     stream.on('close', () => {
+      clearResumeSchedule()
       rateLimitState = null
       this.context.socket.emit(SOCKET_EVENTS.SSH_ERROR, VALIDATION_MESSAGES.CONNECTION_CLOSED)
       this.context.socket.disconnect()

app/types/contracts/v1/socket.ts

@@ -238,7 +238,7 @@ export type AuthenticationEvent =
 // Server → Client events
 export interface ServerToClientEvents {
   // Stream data
-  data: (chunk: string) => void
+  data: (chunk: string | Buffer) => void
   // Auth flow
   authentication: (payload: AuthenticationEvent) => void
   authFailure: (payload: { error: string; method: string }) => void

package-lock.json

@@ -47,7 +47,7 @@
     "socket.io": "^4.8.1",
     "ssh2": "1.17",
     "validator": "^13.15.23",
-    "webssh2_client": "^3.2.0",
+    "webssh2_client": "^3.2.2",
     "zod": "^4.1.12"
   },
   "scripts": {