dont drop user fd until node reads a synthetic EOF (#93)

turns out #51 reduced the race rate
but did not eliminate it completely. we still see the data race bug very
rarely and it is exacerbated on high-numbers of repeats in the tests
and/or high nodejs event loop utilization

my old theory was:
- the user fd is O_NONBLOCK clear so any writes there should block until
its made its way into the 4kb intermediary buffer
- as a result, when the .wait returns, we know that the program output
has at least made it fully into the the user fd input buffer
- because we poll until `controller_inq == 0 && controller_outq == 0 &&
user_inq == 0 && user_outq == 0`, in theory we should never call the js
side exit code until theres truly nothing left in the pipe (i.e. the
readstream on the nodejs side has it in its own buffer)

HOWEVER, with some logging i saw a few cases where in fact
`controller_inq == 0 && controller_outq == 0 && user_inq == 0 &&
user_outq == 0`, the program had exited, yet the nodejs side still
missed some data

what i think is _actually_ happening is that
1. one of libuv's io threads reads the data from the controller side and
queues a data event on the stream
2. polling exits as now theres no more data in the queues we immediately
`drop(user_fd);` which [sets TTY_OTHER_CLOSED
synchronously](https://github.com/torvalds/linux/blob/4ff71af020ae59ae2d83b174646fc2ad9fcd4dc4/drivers/tty/pty.c#L66)
and we get an error event queued on the stream
3. if the nodejs event loop happens to read error before data, we emit
'end' and mark the data as read even though technically nodejs hasnt
processed the data event yet so we drop data :(

how we fix it:
1. axe poll_pty_fds_until_read
2. make Pty struct own user_fd and expose a method for the js side to
drop this fd when its done
3. when child.wait finishes, write a synthetic 'EOF' that is actually a
custom OSC terminal control sequence (`\x1B]7878\x1B\\`, 7878 is RUST on
the phonepad :)) to the user fd (a cursory search shows no results, it
seems _very_ unlikely for this sequence to appear randomly)
4. on the nodejs wrapper side, create a transform stream that parses out
the synthetic EOF and emits it as a custom event when it happens
5. when the nodejs side hits this EOF, we know we are actually at the
end of the data stream and can safely drop user_fd

node-pty had the [same
problem](microsoft/node-pty#72) and did the
:grug: brain thing and [added a wait
250ms](microsoft/vscode@9464b54
) so im calling it slightly more ok

Author: jackyzha0

index.d.ts

@@ -45,6 +45,7 @@ export interface Size {
 }
 export const MAX_U16_VALUE: number;
 export const MIN_U16_VALUE: number;
+export declare function getSyntheticEofSequence(): Buffer;
 /** Resize the terminal. */
 export declare function ptyResize(fd: number, size: Size): void;
 /**
@@ -66,5 +67,10 @@ export declare class Pty {
    * once (it will error the second time). The caller is responsible for closing the file
    * descriptor.
    */
-  takeFd(): c_int;
+  takeControllerFd(): c_int;
+  /**
+   * Closes the owned file descriptor for the PTY controller. The Nodejs side must call this
+   * when it is done with the file descriptor to avoid leaking FDs.
+   */
+  dropUserFd(): void;
 }

index.js

@@ -326,6 +326,7 @@ const {
   Operation,
   MAX_U16_VALUE,
   MIN_U16_VALUE,
+  getSyntheticEofSequence,
   ptyResize,
   setCloseOnExec,
   getCloseOnExec,
@@ -335,6 +336,7 @@ module.exports.Pty = Pty;
 module.exports.Operation = Operation;
 module.exports.MAX_U16_VALUE = MAX_U16_VALUE;
 module.exports.MIN_U16_VALUE = MIN_U16_VALUE;
+module.exports.getSyntheticEofSequence = getSyntheticEofSequence;
 module.exports.ptyResize = ptyResize;
 module.exports.setCloseOnExec = setCloseOnExec;
 module.exports.getCloseOnExec = getCloseOnExec;

npm/darwin-arm64/package.json

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

npm/darwin-x64/package.json

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

npm/linux-x64-gnu/package.json

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

package-lock.json

@@ -1,6 +1,6 @@
 {
   "name": "@replit/ruspty",
-  "version": "3.5.3",
+  "version": "3.6.0",
   "main": "dist/wrapper.js",
   "types": "dist/wrapper.d.ts",
   "author": "Szymon Kaliski <hi@szymonkaliski.com>",
@@ -40,7 +40,7 @@
     "build:wrapper": "tsup",
     "prepublishOnly": "napi prepublish -t npm",
     "test": "vitest run",
-    "test:ci": "vitest --reporter=verbose --reporter=github-actions run",
+    "test:ci": "vitest --reporter=verbose --reporter=github-actions --allowOnly run",
     "test:hang": "vitest run --reporter=hanging-process",
     "universal": "napi universal",
     "version": "napi version",

package.json

@@ -7,17 +7,14 @@ use std::os::fd::{FromRawFd, IntoRawFd, RawFd};
 use std::os::unix::process::CommandExt;
 use std::process::{Command, Stdio};
 use std::thread;
-use std::time::Duration;
 
-use backoff::backoff::Backoff;
-use backoff::ExponentialBackoffBuilder;
-use napi::bindgen_prelude::JsFunction;
+use napi::bindgen_prelude::{Buffer, JsFunction};
 use napi::threadsafe_function::{ErrorStrategy, ThreadsafeFunction, ThreadsafeFunctionCallMode};
 use napi::Status::GenericFailure;
 use napi::{self, Env};
 use nix::errno::Errno;
 use nix::fcntl::{fcntl, FcntlArg, FdFlag, OFlag};
-use nix::libc::{self, c_int, ioctl, FIONREAD, TIOCOUTQ, TIOCSCTTY, TIOCSWINSZ};
+use nix::libc::{self, c_int, TIOCSCTTY, TIOCSWINSZ};
 use nix::pty::{openpty, Winsize};
 use nix::sys::termios::{self, SetArg};
 
@@ -31,6 +28,7 @@ mod sandbox;
 #[allow(dead_code)]
 struct Pty {
   controller_fd: Option<OwnedFd>,
+  user_fd: Option<OwnedFd>,
   /// The pid of the forked process.
   pub pid: u32,
 }
@@ -89,61 +87,15 @@ pub const MAX_U16_VALUE: u16 = u16::MAX;
 #[napi]
 pub const MIN_U16_VALUE: u16 = u16::MIN;
 
-fn cast_to_napi_error(err: Errno) -> napi::Error {
-  napi::Error::new(GenericFailure, err)
-}
-
-// if the child process exits before the controller fd is fully read or the user fd is fully
-// flushed, we might accidentally end in a case where onExit is called but js hasn't had
-// the chance to fully read the controller fd
-// let's wait until the controller fd is fully read before we call onExit
-fn poll_pty_fds_until_read(controller_fd: RawFd, user_fd: RawFd) {
-  let mut backoff = ExponentialBackoffBuilder::default()
-    .with_initial_interval(Duration::from_millis(1))
-    .with_max_interval(Duration::from_millis(100))
-    .with_max_elapsed_time(Some(Duration::from_secs(1)))
-    .build();
-
-  loop {
-    // check both input and output queues for both FDs
-    let mut controller_inq: i32 = 0;
-    let mut controller_outq: i32 = 0;
-    let mut user_inq: i32 = 0;
-    let mut user_outq: i32 = 0;
-
-    // safe because we're passing valid file descriptors and properly sized integers
-    unsafe {
-      // check bytes waiting to be read (FIONREAD, equivalent to TIOCINQ on Linux)
-      if ioctl(controller_fd, FIONREAD, &mut controller_inq) == -1
-        || ioctl(user_fd, FIONREAD, &mut user_inq) == -1
-      {
-        // break if we can't read
-        break;
-      }
-
-      // check bytes waiting to be written (TIOCOUTQ)
-      if ioctl(controller_fd, TIOCOUTQ, &mut controller_outq) == -1
-        || ioctl(user_fd, TIOCOUTQ, &mut user_outq) == -1
-      {
-        // break if we can't read
-        break;
-      }
-    }
+const SYNTHETIC_EOF: &[u8] = b"\x1B]7878\x1B\\";
 
-    // if all queues are empty, we're done
-    if controller_inq == 0 && controller_outq == 0 && user_inq == 0 && user_outq == 0 {
-      break;
-    }
+#[napi]
+pub fn get_synthetic_eof_sequence() -> Buffer {
+  SYNTHETIC_EOF.into()
+}
 
-    // apply backoff strategy
-    if let Some(d) = backoff.next_backoff() {
-      thread::sleep(d);
-      continue;
-    } else {
-      // we have exhausted our attempts
-      break;
-    }
-  }
+fn cast_to_napi_error(err: Errno) -> napi::Error {
+  napi::Error::new(GenericFailure, err)
 }
 
 #[napi]
@@ -347,9 +299,10 @@ impl Pty {
     thread::spawn(move || {
       let wait_result = child.wait();
 
-      // try to wait for the controller fd to be fully read
-      poll_pty_fds_until_read(raw_controller_fd, raw_user_fd);
-      drop(user_fd);
+      // by this point, child has closed its copy of the user_fd
+      // lets inject our synthetic EOF OSC into the user_fd
+      // its ok to ignore the result here as we have a timeout on the nodejs side to handle if this write fails
+      let _ = write_syn_eof_to_fd(raw_user_fd);
 
       match wait_result {
         Ok(status) => {
@@ -379,6 +332,7 @@ impl Pty {
 
     Ok(Pty {
       controller_fd: Some(controller_fd),
+      user_fd: Some(user_fd),
       pid,
     })
   }
@@ -388,7 +342,7 @@ impl Pty {
   /// descriptor.
   #[napi]
   #[allow(dead_code)]
-  pub fn take_fd(&mut self) -> Result<c_int, napi::Error> {
+  pub fn take_controller_fd(&mut self) -> Result<c_int, napi::Error> {
     if let Some(fd) = self.controller_fd.take() {
       Ok(fd.into_raw_fd())
     } else {
@@ -398,6 +352,15 @@ impl Pty {
       ))
     }
   }
+
+  /// Closes the owned file descriptor for the PTY controller. The Nodejs side must call this
+  /// when it is done with the file descriptor to avoid leaking FDs.
+  #[napi]
+  #[allow(dead_code)]
+  pub fn drop_user_fd(&mut self) -> Result<(), napi::Error> {
+    self.user_fd.take();
+    Ok(())
+  }
 }
 
 /// Resize the terminal.
@@ -492,3 +455,35 @@ fn set_nonblocking(fd: i32) -> Result<(), napi::Error> {
   }
   Ok(())
 }
+
+fn write_syn_eof_to_fd(fd: libc::c_int) -> std::io::Result<()> {
+  let mut remaining = SYNTHETIC_EOF;
+  while !remaining.is_empty() {
+    match unsafe {
+      libc::write(
+        fd,
+        remaining.as_ptr() as *const libc::c_void,
+        remaining.len(),
+      )
+    } {
+      -1 => {
+        let err = std::io::Error::last_os_error();
+        if err.kind() == std::io::ErrorKind::Interrupted {
+          continue;
+        }
+
+        return Err(err);
+      }
+      0 => {
+        return Err(std::io::Error::new(
+          std::io::ErrorKind::WriteZero,
+          "write returned 0",
+        ));
+      }
+      n => {
+        remaining = &remaining[n as usize..];
+      }
+    }
+  }
+  Ok(())
+}

src/lib.rs

@@ -0,0 +1,85 @@
+import { Transform } from 'node:stream';
+import { getSyntheticEofSequence } from './index.js';
+
+// keep in sync with lib.rs::SYNTHETIC_EOF
+export const SYNTHETIC_EOF = getSyntheticEofSequence();
+export const EOF_EVENT = 'synthetic-eof';
+
+// get the longest suffix of buffer that is a prefix of SYNTHETIC_EOF
+function getBufferEndPrefixLength(buffer: Buffer) {
+  const maxLen = Math.min(buffer.length, SYNTHETIC_EOF.length);
+  for (let len = maxLen; len > 0; len--) {
+    let match = true;
+    for (let i = 0; i < len; i++) {
+      if (buffer[buffer.length - len + i] !== SYNTHETIC_EOF[i]) {
+        match = false;
+        break;
+      }
+    }
+
+    if (match) {
+      return len;
+    }
+  }
+
+  return 0;
+}
+
+export class SyntheticEOFDetector extends Transform {
+  buffer: Buffer;
+
+  constructor(options = {}) {
+    super(options);
+    this.buffer = Buffer.alloc(0);
+  }
+
+  _transform(chunk: Buffer, _encoding: string, callback: () => void) {
+    const searchData = Buffer.concat([this.buffer, chunk]);
+    const eofIndex = searchData.indexOf(SYNTHETIC_EOF);
+
+    if (eofIndex !== -1) {
+      // found EOF - emit everything before it
+      if (eofIndex > 0) {
+        this.push(searchData.subarray(0, eofIndex));
+      }
+
+      this.emit(EOF_EVENT);
+
+      // emit everything after EOF (if any) and clear buffer
+      const afterEOF = searchData.subarray(eofIndex + SYNTHETIC_EOF.length);
+      if (afterEOF.length > 0) {
+        this.push(afterEOF);
+      }
+
+      this.buffer = Buffer.alloc(0);
+    } else {
+      // no EOF - buffer potential partial match at end
+
+      // get the longest suffix of buffer that is a prefix of SYNTHETIC_EOF
+      // and emit everything before it
+      // this is done for the case which the eof happened to be split across multiple chunks
+      const commonPrefixLen = getBufferEndPrefixLength(searchData);
+
+      if (commonPrefixLen > 0) {
+        const emitSize = searchData.length - commonPrefixLen;
+        if (emitSize > 0) {
+          this.push(searchData.subarray(0, emitSize));
+        }
+        this.buffer = searchData.subarray(emitSize);
+      } else {
+        this.push(searchData);
+        this.buffer = Buffer.alloc(0);
+      }
+    }
+
+    callback();
+  }
+
+  _flush(callback: () => void) {
+    if (this.buffer.length > 0) {
+      this.push(this.buffer);
+    }
+
+    callback();
+  }
+}