3.3.0 remove a layer of indirection (unnecessary passthrough) (#49)

we shouldnt use passthrough if we dont need it, we got wrecked by the
default watermark on the passthrough stream

npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@replit/ruspty-darwin-arm64",
-  "version": "3.2.4",
+  "version": "3.3.0",
   "os": [
     "darwin"
   ],

npm/darwin-x64/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@replit/ruspty-darwin-x64",
-  "version": "3.2.4",
+  "version": "3.3.0",
   "os": [
     "darwin"
   ],

npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@replit/ruspty-linux-x64-gnu",
-  "version": "3.2.4",
+  "version": "3.3.0",
   "os": [
     "linux"
   ],

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@replit/ruspty",
-  "version": "3.2.4",
+  "version": "3.3.0",
   "main": "dist/wrapper.js",
   "types": "dist/wrapper.d.ts",
   "author": "Szymon Kaliski <[email protected]>",

tests/index.test.ts

@@ -76,7 +76,7 @@ describe(
     test('captures an exit code', () =>
       new Promise<void>((done) => {
         const oldFds = getOpenFds();
-        new Pty({
+        const pty = new Pty({
           command: '/bin/sh',
           args: ['-c', 'exit 17'],
           onExit: (err, exitCode) => {
@@ -86,6 +86,9 @@ describe(
             done();
           },
         });
+
+        // set a pty reader so it can flow
+        pty.read.on('data', () => { });
       }));
 
     test('can be written to', () =>
@@ -272,6 +275,8 @@ describe(
           }
         },
       });
+
+      pty.read.on('data', () => { });
     }));
 
     test('resize after close shouldn\'t throw', () => new Promise<void>((done, reject) => {
@@ -287,6 +292,8 @@ describe(
         },
       });
 
+      pty.read.on('data', () => { });
+
       pty.close();
       expect(() => {
         pty.resize({ rows: 60, cols: 100 });
@@ -305,13 +312,13 @@ describe(
             command: '/bin/sh',
             args: [
               '-c',
-              `for i in $(seq 0 ${n}); do /bin/echo $i; done && exit`,
+              'seq 0 1024'
             ],
             onExit: (err, exitCode) => {
               expect(err).toBeNull();
               expect(exitCode).toBe(0);
-              expect(buffer.toString().trim()).toBe(
-                [...Array(n + 1).keys()].join('\r\n'),
+              expect(buffer.toString().trim().split('\n').map(Number)).toStrictEqual(
+                Array.from({ length: n + 1 }, (_, i) => i),
               );
               expect(getOpenFds()).toStrictEqual(oldFds);
               done();
@@ -325,9 +332,35 @@ describe(
         }),
     );
 
+    test('doesnt miss large output from fast commands',
+      () =>
+        new Promise<void>((done) => {
+          const payload = `hello`.repeat(4096);
+          let buffer = Buffer.from('');
+          const pty = new Pty({
+            command: '/bin/echo',
+            args: [
+              '-n',
+              payload
+            ],
+            onExit: (err, exitCode) => {
+              expect(err).toBeNull();
+              expect(exitCode).toBe(0);
+              // account for the newline
+              expect(buffer.toString().length).toBe(payload.length);
+              done();
+            },
+          });
+
+          const readStream = pty.read;
+          readStream.on('data', (data) => {
+            buffer = Buffer.concat([buffer, data]);
+          });
+        })
+    );
+
     testSkipOnDarwin(
       'does not leak files',
-      { repeats: 4 },
       () =>
         new Promise<void>((done) => {
           const oldFds = getOpenFds();
@@ -373,7 +406,6 @@ describe(
 
     test(
       'can run concurrent shells',
-      { repeats: 4 },
       () =>
         new Promise<void>((done) => {
           const oldFds = getOpenFds();

wrapper.ts

@@ -1,4 +1,4 @@
-import { PassThrough, type Readable, type Writable } from 'node:stream';
+import { type Readable, Writable } from 'node:stream';
 import { ReadStream } from 'node:tty';
 import {
   Pty as RawPty,
@@ -45,21 +45,30 @@ type ExitResult = {
 export class Pty {
   #pty: RawPty;
   #fd: number;
-  #fdEnded: boolean = false;
+
+  #handledClose: boolean = false;
+  #handledEndOfData: boolean = false;
+
   #socket: ReadStream;
+  get read(): Readable {
+    return this.#socket;
+  }
 
-  read: Readable;
   write: Writable;
 
   constructor(options: PtyOptions) {
     const realExit = options.onExit;
 
-    let resolve: (value: ExitResult) => void;
-    let exitResult: Promise<ExitResult> = new Promise((res) => {
-      resolve = res;
+    let markExited: (value: ExitResult) => void;
+    let exitResult: Promise<ExitResult> = new Promise((resolve) => {
+      markExited = resolve;
+    });
+    let markFdClosed: () => void;
+    let fdClosed = new Promise<void>((resolve) => {
+      markFdClosed = resolve;
     });
     const mockedExit = (error: NodeJS.ErrnoException | null, code: number) => {
-      resolve({ error, code });
+      markExited({ error, code });
     };
 
     // when pty exits, we should wait until the fd actually ends (end OR error)
@@ -70,27 +79,29 @@ export class Pty {
     // Transfer ownership of the FD to us.
     this.#fd = this.#pty.takeFd();
 
-    this.#socket = new ReadStream(this.#fd);
-    const userFacingRead = new PassThrough();
-    const userFacingWrite = new PassThrough();
-    this.#socket.pipe(userFacingRead);
-    userFacingWrite.pipe(this.#socket);
-    this.read = userFacingRead;
-    this.write = userFacingWrite;
+    this.#socket = new ReadStream(this.#fd)
+    this.write = new Writable({
+      write: this.#socket.write.bind(this.#socket),
+    });
 
     // catch end events
-    const handleClose = () => {
-      if (this.#fdEnded) {
+    const handleEnd = async () => {
+      if (this.#handledEndOfData) {
         return;
       }
 
-      this.#fdEnded = true;
-      exitResult.then((result) => {
-        realExit(result.error, result.code)
-      });
-      userFacingRead.end();
-    };
-    this.#socket.on('close', handleClose);
+      this.#handledEndOfData = true;
+
+      // must wait for fd close and exit result before calling real exit
+      await fdClosed;
+      const result = await exitResult;
+      realExit(result.error, result.code)
+    }
+
+    this.read.on('end', handleEnd);
+    this.read.on('close', () => {
+      markFdClosed();
+    });
 
     // PTYs signal their done-ness with an EIO error. we therefore need to filter them out (as well as
     // cleaning up other spurious errors) so that the user doesn't need to handle them and be in
@@ -108,25 +119,26 @@ export class Pty {
           // EIO only happens when the child dies. It is therefore our only true signal that there
           // is nothing left to read and we can start tearing things down. If we hadn't received an
           // error so far, we are considered to be in good standing.
-          this.#socket.off('error', handleError);
-          this.#socket.end();
+          this.read.off('error', handleError);
+          handleEnd();
           return;
         }
       }
-
-      this.read.emit('error', err);
     };
-    this.#socket.on('error', handleError);
+
+    this.read.on('error', handleError);
   }
 
   close() {
+    this.#handledClose = true;
+
     // end instead of destroy so that the user can read the last bits of data
     // and allow graceful close event to mark the fd as ended
     this.#socket.end();
   }
 
   resize(size: Size) {
-    if (this.#fdEnded) {
+    if (this.#handledClose || this.#handledEndOfData) {
       return;
     }