fix(unix): pace the EAGAIN write retry so a stalled reader can't saturate the thread

src/unixTerminal.test.ts

@@ -124,6 +124,67 @@ if (process.platform !== 'win32') {
         term.destroy();
       });
     });
+    describe('EAGAIN backpressure', () => {
+      let realWrite: typeof fs.write;
+
+      beforeEach(() => {
+        realWrite = fs.write;
+      });
+
+      afterEach(() => {
+        (fs as any).write = realWrite;
+      });
+
+      it('should pace retries when the fd keeps returning EAGAIN', (done) => {
+        const term = new UnixTerminal('node', [ '-e', 'setInterval(() => {}, 1000)' ]);
+        let attempts = 0;
+        (fs as any).write = (fd: number, buf: Buffer, offset: number, cb: Function): void => {
+          attempts++;
+          const err: any = new Error('EAGAIN');
+          err.code = 'EAGAIN';
+          process.nextTick(() => cb(err));
+        };
+
+        term.write('trigger');
+        setTimeout(() => {
+          (fs as any).write = realWrite;
+          term.destroy();
+          // At 5ms pacing this is ~20 attempts in 100ms. An immediate re-attempt
+          // produces tens of thousands, which is the busy-loop being fixed.
+          assert.ok(attempts > 0, 'expected the EAGAIN path to be exercised');
+          assert.ok(attempts < 100, `expected paced retries, got ${attempts} attempts in 100ms`);
+          done();
+        }, 100);
+      });
+
+      it('should not write after dispose while a write is in flight', (done) => {
+        const term = new UnixTerminal('node', [ '-e', 'setInterval(() => {}, 1000)' ]);
+        let attempts = 0;
+        let release: (() => void) | undefined;
+        (fs as any).write = (fd: number, buf: Buffer, offset: number, cb: Function): void => {
+          attempts++;
+          const err: any = new Error('EAGAIN');
+          err.code = 'EAGAIN';
+          // Hold the completion so it lands *after* dispose, as a real in-flight
+          // write on the libuv threadpool would.
+          release = () => cb(err);
+        };
+
+        term.write('trigger');
+        setTimeout(() => {
+          const before = attempts;
+          term.destroy();
+          release!();
+          setTimeout(() => {
+            (fs as any).write = realWrite;
+            assert.strictEqual(attempts, before,
+              'a write was issued after dispose; the in-flight completion re-armed the retry');
+            done();
+          }, 50);
+        }, 20);
+      });
+    });
+
     describe('signals in parent and child', () => {
       it('SIGINT - custom in parent and child', done => {
         // this test is cumbersome - we have to run it in a sub process to

src/unixTerminal.ts

@@ -308,6 +308,13 @@ interface IWriteTask {
   offset: number;
 }
 
+/**
+ * How long to wait before re-attempting a write that returned EAGAIN. Large enough to
+ * stop the retry from monopolising the thread while a reader is stalled, small enough
+ * to be invisible to interactive writes.
+ */
+const EAGAIN_RETRY_DELAY_MS = 5;
+
 /**
  * A custom write stream that writes directly to a file descriptor with proper
  * handling of backpressure and errors. This avoids some event loop exhaustion
@@ -316,7 +323,8 @@ interface IWriteTask {
 class CustomWriteStream implements IDisposable {
 
   private readonly _writeQueue: IWriteTask[] = [];
-  private _writeImmediate: NodeJS.Immediate | undefined;
+  private _writeRetry: NodeJS.Timeout | undefined;
+  private _isDisposed: boolean = false;
 
   constructor(
     private readonly _fd: number,
@@ -325,8 +333,9 @@ class CustomWriteStream implements IDisposable {
   }
 
   dispose(): void {
-    clearImmediate(this._writeImmediate);
-    this._writeImmediate = undefined;
+    this._isDisposed = true;
+    clearTimeout(this._writeRetry);
+    this._writeRetry = undefined;
   }
 
   write(data: string | Buffer): void {
@@ -345,7 +354,11 @@ class CustomWriteStream implements IDisposable {
   }
 
   private _processWriteQueue(): void {
-    this._writeImmediate = undefined;
+    if (this._isDisposed) {
+      return;
+    }
+
+    this._writeRetry = undefined;
 
     if (this._writeQueue.length === 0) {
       return;
@@ -357,11 +370,20 @@ class CustomWriteStream implements IDisposable {
     // than using the `net.Socket`/`tty.WriteStream` wrappers which swallow and
     // mask errors like EAGAIN and can cause the thread to block indefinitely.
     fs.write(this._fd, task.buffer, task.offset, (err, written) => {
+      // The fd may have been closed while this write was in flight. Anything below
+      // would either re-arm the retry or write again against a dead descriptor.
+      if (this._isDisposed) {
+        return;
+      }
+
       if (err) {
         if ('code' in err && err.code === 'EAGAIN') {
-          // `setImmediate` is used to yield to the event loop and re-attempt
-          // the write later.
-          this._writeImmediate = setImmediate(() => this._processWriteQueue());
+          // Re-attempt the write later. A delay rather than `setImmediate`: EAGAIN here
+          // means the reader has stopped draining the pty, so the branch keeps failing
+          // and an immediate re-attempt turns it into a busy-loop that saturates the
+          // thread. A reader that is merely slow applies backpressure without reaching
+          // this branch at all, so the delay does not throttle normal writes.
+          this._writeRetry = setTimeout(() => this._processWriteQueue(), EAGAIN_RETRY_DELAY_MS);
         } else {
           // Stop processing immediately on unexpected error and log
           this._writeQueue.length = 0;