fix(usbmux): handle decoder errors and preserve unconsumed bytes on connect

Author: harsha509

src/lib/usbmux/index.ts

@@ -6,7 +6,7 @@ import {BaseSocketService} from '../../base-socket-service.js';
 import {getLogger} from '../logger.js';
 import {type PairRecord, processPlistResponse} from '../pair-record/index.js';
 import {type RawPairRecordResponse} from '../pair-record/pair-record.js';
-import {LengthBasedSplitter, parsePlist} from '../plist/index.js';
+import {parsePlist} from '../plist/index.js';
 import type {PlistDictionary} from '../types.js';
 import {type DecodedUsbmux, UsbmuxDecoder} from '../usbmux/usbmux-decoder.js';
 import {UsbmuxEncoder} from '../usbmux/usbmux-encoder.js';
@@ -39,7 +39,6 @@ const log = getLogger('Usbmux');
 export const USBMUXD_PORT = 27015;
 export const DEFAULT_USBMUXD_SOCKET = '/var/run/usbmuxd';
 export const DEFAULT_USBMUXD_HOST = '127.0.0.1';
-export const MAX_FRAME_SIZE = 100 * 1024 * 1024; // 1MB
 
 // Result codes from usbmuxd
 export const USBMUX_RESULT = {
@@ -68,7 +67,6 @@ export interface SocketOptions {
  */
 export class Usbmux extends BaseSocketService {
   private readonly _decoder: UsbmuxDecoder;
-  private readonly _splitter: LengthBasedSplitter;
   private readonly _encoder: UsbmuxEncoder;
   private _tag: number;
   private readonly _responseCallbacks: Record<number, (data: DecodedUsbmux) => void>;
@@ -81,16 +79,10 @@ export class Usbmux extends BaseSocketService {
     super(socketClient);
 
     this._decoder = new UsbmuxDecoder();
-    this._splitter = new LengthBasedSplitter({
-      readableStream: socketClient,
-      littleEndian: true,
-      maxFrameLength: MAX_FRAME_SIZE,
-      lengthFieldOffset: 0,
-      lengthFieldLength: 4,
-      lengthAdjustment: 0,
-    });
-
     this._socketClient.pipe(this._decoder);
+    this._decoder.on('error', (err: Error) => {
+      log.error(`Usbmux decoder error: ${err.message}`);
+    });
 
     this._encoder = new UsbmuxEncoder();
     this._encoder.pipe(this._socketClient);
@@ -219,20 +211,17 @@ export class Usbmux extends BaseSocketService {
 
       if (data.payload.Number === USBMUX_RESULT.OK) {
         // Detach constructor-owned consumers from the raw socket so the caller
-        // gets full byte-stream ownership. Leaving the splitter/decoder attached
-        // lets their unread buffers grow, eventually applying backpressure that
-        // stalls long-lived forwards.
-        //
-        // Order matters: unpipe() must happen before removeAllListeners() /
-        // shutdown(). Removing listeners first breaks Node's pipe cleanup, leaving
-        // orphaned source listeners that continue pushing bytes into detached
-        // streams and eventually stall the socket.
-        this._socketClient.unpipe(this._splitter);
-        this._splitter.unpipe(this._decoder);
+        // gets full byte-stream ownership.
         this._socketClient.unpipe(this._decoder);
         this._encoder.unpipe(this._socketClient);
-        this._splitter.shutdown();
         this._decoder.removeAllListeners('data');
+
+        // Hand any unconsumed bytes back to the raw socket stream so the caller
+        // receives them immediately when attaching listeners.
+        const pending = this._decoder.buffer;
+        if (pending.length > 0) {
+          this._socketClient.unshift(pending);
+        }
         return this._socketClient;
       } else if (data.payload.Number === USBMUX_RESULT.CONNREFUSED) {
         throw new Error(`Connection was refused to port ${port}`);

src/lib/usbmux/usbmux-decoder.ts

@@ -24,6 +24,10 @@ export class UsbmuxDecoder extends Transform {
     super({objectMode: true});
   }
 
+  get buffer(): Buffer {
+    return this._buffer;
+  }
+
   _transform(chunk: Buffer, encoding: BufferEncoding, callback: TransformCallback): void {
     // Append the new chunk to the internal buffer
     this._buffer = Buffer.concat([this._buffer, chunk]);
@@ -38,12 +42,16 @@ export class UsbmuxDecoder extends Transform {
         break; // Wait for more data
       }
 
-      // Extract the full message
-      const message = this._buffer.slice(0, totalLength);
-      this._decode(message);
+      // Extract the full message and remove it from the buffer before decoding,
+      // so any synchronous data listeners see only the unconsumed remainder.
+      const message = this._buffer.subarray(0, totalLength);
+      this._buffer = this._buffer.subarray(totalLength);
 
-      // Remove the processed message from the buffer
-      this._buffer = this._buffer.slice(totalLength);
+      try {
+        this._decode(message);
+      } catch (err) {
+        return callback(err instanceof Error ? err : new Error(String(err)));
+      }
     }
     callback();
   }
@@ -56,7 +64,7 @@ export class UsbmuxDecoder extends Transform {
       tag: data.readUInt32LE(12),
     };
 
-    const payload = data.slice(HEADER_LENGTH);
+    const payload = data.subarray(HEADER_LENGTH);
     this.push({header, payload: parsePlist(payload)} as DecodedUsbmux);
   }
 }

test/unit/usbmux/usbmux.spec.ts

@@ -1,11 +1,18 @@
 import assert from 'node:assert/strict';
-import {type Server, type Socket} from 'node:net';
+import {type AddressInfo, type Server, type Socket, createConnection, createServer} from 'node:net';
+import {resolve} from 'node:path';
 import {afterEach, beforeEach, describe, it} from 'node:test';
+import {fileURLToPath} from 'node:url';
+
+import {fs, node} from '@appium/support';
 
 import {type Device, Usbmux} from '../../../src/lib/usbmux/index.js';
+import {UsbmuxDecoder} from '../../../src/lib/usbmux/usbmux-decoder.js';
 import {prioritizeUsbOverNetworkForDuplicateUdids} from '../../../src/lib/usbmux/utils.js';
 import {UDID, fixtures, getServerWithFixtures} from '../fixtures/index.js';
 
+const PKG_ROOT = node.getModuleRootSync('appium-ios-remotexpc', fileURLToPath(import.meta.url));
+
 const DUP_UDID = 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa';
 
 function mockUsbmuxDevice(
@@ -95,6 +102,60 @@ describe('usbmux', function () {
     }
   });
 
+  it('should preserve remainder bytes in decoder buffer when partial chunk arrives', function () {
+    const decoder = new UsbmuxDecoder();
+    const chunk = Buffer.from([0x05, 0x00, 0x00, 0x00]);
+    decoder.write(chunk);
+    assert.deepStrictEqual(decoder.buffer, chunk);
+  });
+
+  it('should emit error on malformed payload instead of throwing synchronously in decoder', async function () {
+    const decoder = new UsbmuxDecoder();
+    const errorPromise = new Promise<Error>((resolve) => {
+      decoder.once('error', resolve);
+    });
+
+    const header = Buffer.alloc(16);
+    header.writeUInt32LE(20, 0); // length: 20 bytes total (16 header + 4 payload)
+    header.writeUInt32LE(1, 4); // version
+    header.writeUInt32LE(8, 8); // type
+    header.writeUInt32LE(1, 12); // tag
+    const invalidPayload = Buffer.from('bad!');
+
+    decoder.write(Buffer.concat([header, invalidPayload]));
+    const err = await errorPromise;
+    assert.ok(err instanceof Error);
+  });
+
+  it('should unshift unconsumed trailing bytes received alongside connect result', async function () {
+    const connectFixture = Buffer.from(
+      await fs.readFile(resolve(PKG_ROOT, 'test', 'unit', 'fixtures', 'usbmuxconnectmessage.bin')),
+    );
+    const trailingGreeting = Buffer.from('REMOTE_SERVICE_GREETING');
+
+    server = createServer((s: Socket) => {
+      s.once('data', (data: Buffer) => {
+        const clientTag = data.readUInt32LE(12);
+        connectFixture.writeUInt32LE(clientTag, 12);
+        s.write(Buffer.concat([connectFixture, trailingGreeting]));
+      });
+    });
+    server.listen();
+    const addr = server.address() as AddressInfo;
+    socket = createConnection(addr.port);
+
+    usbmux = new Usbmux(socket);
+    const connectedSocket = await usbmux.connect(1, 62078);
+
+    const received = await new Promise<Buffer>((resolve) => {
+      connectedSocket.once('data', resolve);
+      connectedSocket.resume();
+    });
+
+    assert.deepStrictEqual(received, trailingGreeting);
+    connectedSocket.destroy();
+  });
+
   it('should order duplicate UDIDs with USB before Network', function () {
     const net = mockUsbmuxDevice(2, DUP_UDID, 'Network');
     const usb = mockUsbmuxDevice(1, DUP_UDID, 'USB');