Add opt-in flow control: --flow-control {backpressure,discard} (#631)

proto/ETerminal.proto

@@ -57,10 +57,25 @@ message PortForwardData {
   optional bool closed = 5;
 }
 
+enum FlowControlMode {
+  // Legacy behavior: no application-level buffering or kernel buffer
+  // tuning. This is the default and what servers assume for old clients
+  // that don't send flow_control_mode: current users are unchanged.
+  FLOW_CONTROL_NONE = 0;
+  // Bounded buffering; when full, stop reading from the source. Lossless
+  // (like plain ssh, but with small, tuned buffers so Ctrl-C stays
+  // responsive on a saturated link).
+  FLOW_CONTROL_BACKPRESSURE = 1;
+  // Bounded buffering; when full, drop the oldest pending output. The
+  // producing process never stalls and the display stays near real time.
+  FLOW_CONTROL_DISCARD = 2;
+}
+
 message InitialPayload {
   optional bool jumphost = 1 [default = false];
   repeated PortForwardSourceRequest reversetunnels = 2;
   map<string, string> environmentvariables = 3;
+  optional FlowControlMode flow_control_mode = 4 [default = FLOW_CONTROL_NONE];
 }
 
 message InitialResponse {
@@ -75,6 +90,9 @@ message ConfigParams {
 message TermInit {
   repeated string environmentnames = 1;
   repeated string environmentvalues = 2;
+  // Tells etterminal whether the attached client opted into flow control,
+  // so it can shrink the kernel buffer on its socket to the server.
+  optional FlowControlMode flow_control_mode = 3 [default = FLOW_CONTROL_NONE];
 }
 
 message TerminalUserInfo {

src/base/Headers.hpp

@@ -367,6 +367,12 @@ inline bool waitOnSocketData(int fd) {
     if (errno == EINTR) {
       // Interrupted by the signal, the caller will retry.
       return false;
+    } else if (errno == EBADF || errno == EINVAL) {
+      // The fd was closed under us (e.g. peer disconnected). It will never
+      // become readable, and callers use this helper in retry loops, so
+      // returning false would spin at 100% CPU. Throw so the loop treats
+      // it as socket death.
+      throw std::runtime_error("Socket closed while waiting for data");
     } else {
       FATAL_FAIL(selectResult);
     }
@@ -389,8 +395,12 @@ inline bool isSocketWritable(int fd) {
   tv.tv_usec = 0;
   const int selectResult = select(fd + 1, NULL, &fdset, NULL, &tv);
   if (selectResult < 0) {
-    if (errno == EINTR) {
-      // Interrupted by the signal, the caller will retry.
+    if (errno == EINTR || errno == EBADF || errno == EINVAL) {
+      // EINTR: interrupted by a signal. EBADF/EINVAL: the fd was closed
+      // under us (e.g. client disconnected). Returning false is safe in
+      // both cases: callers use this as a non-blocking "can I write more?"
+      // poll and stop on false; a dead socket is then surfaced by the next
+      // read/write on the connection.
       return false;
     } else {
       FATAL_FAIL(selectResult);

src/base/PipeSocketHandler.cpp

@@ -144,4 +144,18 @@ void PipeSocketHandler::stopListening(const SocketEndpoint& endpoint) {
   FATAL_FAIL(::close(sockFd));
 #endif
 }
+
+void PipeSocketHandler::minimizeKernelBuffering(int fd) {
+#ifndef WIN32
+  // Bound the kernel buffer on this unix socket. Terminal output waits
+  // here when the server applies backpressure (stops reading); the default
+  // (~200KB on Linux) is seconds of stale output on a slow link. 64KB does
+  // not limit throughput on a local socket.
+  int sndbuf = 64 * 1024;
+  if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char*)&sndbuf, sizeof(sndbuf)) <
+      0) {
+    LOG(WARNING) << "Failed to set SO_SNDBUF: " << strerror(errno);
+  }
+#endif
+}
 }  // namespace et

src/base/PipeSocketHandler.hpp

@@ -30,6 +30,8 @@ class PipeSocketHandler : public UnixSocketHandler {
    */
   virtual void stopListening(const SocketEndpoint& endpoint);
 
+  virtual void minimizeKernelBuffering(int fd);
+
  protected:
   /** @brief Tracks path -> listening socket descriptors for each pipe. */
   map<string, set<int>> pipeServerSockets;

src/base/SocketHandler.hpp

@@ -20,6 +20,15 @@ class SocketHandler {
    * descriptor.
    */
   virtual bool hasData(int fd) = 0;
+
+  /**
+   * @brief Shrinks the kernel's outgoing queue on fd so that pending data
+   * waits in application-level buffers, where flow control can apply
+   * backpressure or discard stale output. Only called for sessions that
+   * opt into flow control (--flow-control backpressure|discard); the
+   * default is a no-op so legacy sessions keep stock kernel buffering.
+   */
+  virtual void minimizeKernelBuffering(int fd) {}
   /**
    * @brief Reads up to count bytes from fd.
    */

src/base/TcpSocketHandler.cpp

@@ -293,4 +293,23 @@ void TcpSocketHandler::initSocket(int fd) {
         fd, SOL_SOCKET, SO_LINGER, (const char*)&so_linger, sizeof so_linger));
   }
 }
+
+void TcpSocketHandler::minimizeKernelBuffering(int fd) {
+#ifdef TCP_NOTSENT_LOWAT
+  // Keep the kernel's not-yet-sent queue small so pending terminal output
+  // stays in application-level buffers, where flow control can apply
+  // backpressure or discard stale data. Without this, select()/poll()
+  // report the socket writable whenever the autotuned (multi-MB) send
+  // buffer has room, and on a slow link many seconds of stale output pile
+  // up in the kernel where they can no longer be dropped (issue #631).
+  // TCP_NOTSENT_LOWAT only limits *unsent* data; sent-but-unacked
+  // (in-flight) data is unaffected, so throughput on high-BDP links is
+  // preserved. Best effort: not supported on all platforms.
+  int lowat = 32 * 1024;
+  if (setsockopt(fd, IPPROTO_TCP, TCP_NOTSENT_LOWAT, (char*)&lowat,
+                 sizeof(lowat)) < 0) {
+    LOG(WARNING) << "Failed to set TCP_NOTSENT_LOWAT: " << strerror(errno);
+  }
+#endif
+}
 }  // namespace et

src/base/TcpSocketHandler.hpp

@@ -31,6 +31,8 @@ class TcpSocketHandler : public UnixSocketHandler {
    */
   virtual void stopListening(const SocketEndpoint& endpoint);
 
+  virtual void minimizeKernelBuffering(int fd);
+
  protected:
   /** @brief Tracks all listening sockets created per TCP port. */
   map<int, set<int>> portServerSockets;

src/base/WriteBuffer.hpp

@@ -0,0 +1,150 @@
+#ifndef __ET_WRITE_BUFFER__
+#define __ET_WRITE_BUFFER__
+
+#include "Headers.hpp"
+
+namespace et {
+
+enum class WriteBufferMode {
+  BACKPRESSURE,  // Stop accepting data when buffer is full
+  DISCARD        // Drop oldest data when buffer is full
+};
+
+/**
+ * @brief Bounded buffer for pending write data, enabling flow control.
+ *
+ * Supports two modes:
+ * - BACKPRESSURE: canAcceptMore() returns false when full, causing callers
+ *   to stop reading from the source. Processes stall when consumer is slow.
+ * - DISCARD: canAcceptMore() always returns true. When the buffer exceeds
+ *   MAX_BUFFER_SIZE, oldest data is dropped. Processes never stall.
+ */
+class WriteBuffer {
+ public:
+  /** @brief Maximum bytes to buffer before applying backpressure or
+   * discarding. Every byte held here (or in any buffer downstream) is stale
+   * data the user must wait through on a slow link, so keep this small: at
+   * 100KB/s, 64KB is ~0.6s of output. It only needs to absorb short bursts
+   * between drain opportunities (reads are 16KB), not smooth over the
+   * network. This is a soft bound: canAcceptMore() admits a whole chunk
+   * whenever size() is below the limit, so the buffer can exceed it by up
+   * to one chunk. */
+  static constexpr size_t MAX_BUFFER_SIZE = 64 * 1024;  // 64KB
+
+  explicit WriteBuffer(WriteBufferMode mode = WriteBufferMode::BACKPRESSURE)
+      : mode(mode), totalBytes(0), writeOffset(0) {}
+
+  /**
+   * @brief Returns true if the buffer has room for more data.
+   * In BACKPRESSURE mode, returns false when buffer >= MAX_BUFFER_SIZE.
+   * In DISCARD mode, always returns true (old data will be dropped).
+   */
+  bool canAcceptMore() const {
+    if (mode == WriteBufferMode::DISCARD) return true;
+    return totalBytes < MAX_BUFFER_SIZE;
+  }
+
+  /**
+   * @brief Returns true if there is data waiting to be written.
+   */
+  bool hasPendingData() const { return !pending.empty(); }
+
+  /**
+   * @brief Returns the current amount of buffered data in bytes.
+   */
+  size_t size() const { return totalBytes; }
+
+  /**
+   * @brief Returns the buffer mode.
+   */
+  WriteBufferMode getMode() const { return mode; }
+
+  /**
+   * @brief Adds data to the end of the buffer.
+   * In DISCARD mode, drops oldest chunks if the buffer would exceed
+   * MAX_BUFFER_SIZE.
+   * @param data The data to enqueue.
+   */
+  void enqueue(const string& data) {
+    if (data.empty()) return;
+    pending.push_back(data);
+    totalBytes += data.size();
+
+    if (mode == WriteBufferMode::DISCARD) {
+      discardOldest();
+    }
+  }
+
+  /**
+   * @brief Returns a pointer to the next bytes to write and the count.
+   * @param count Output: number of bytes available for writing.
+   * @return Pointer to the data, or nullptr if buffer is empty.
+   */
+  const char* peekData(size_t* count) const {
+    if (pending.empty()) {
+      *count = 0;
+      return nullptr;
+    }
+    const string& front = pending.front();
+    *count = front.size() - writeOffset;
+    return front.data() + writeOffset;
+  }
+
+  /**
+   * @brief Removes bytesWritten from the front of the buffer.
+   * @param bytesWritten Number of bytes successfully written to socket.
+   */
+  void consume(size_t bytesWritten) {
+    if (bytesWritten == 0) return;
+
+    while (bytesWritten > 0 && !pending.empty()) {
+      string& front = pending.front();
+      size_t available = front.size() - writeOffset;
+
+      if (bytesWritten >= available) {
+        // Consumed the entire front chunk
+        bytesWritten -= available;
+        totalBytes -= available;
+        writeOffset = 0;
+        pending.pop_front();
+      } else {
+        // Partial consumption
+        writeOffset += bytesWritten;
+        totalBytes -= bytesWritten;
+        bytesWritten = 0;
+      }
+    }
+  }
+
+  /**
+   * @brief Clears all pending data.
+   */
+  void clear() {
+    pending.clear();
+    totalBytes = 0;
+    writeOffset = 0;
+  }
+
+ private:
+  /**
+   * @brief Drops oldest chunks until buffer is within MAX_BUFFER_SIZE.
+   * Only called in DISCARD mode after enqueue.
+   */
+  void discardOldest() {
+    while (totalBytes > MAX_BUFFER_SIZE && !pending.empty()) {
+      string& front = pending.front();
+      size_t frontSize = front.size() - writeOffset;
+      totalBytes -= frontSize;
+      writeOffset = 0;
+      pending.pop_front();
+    }
+  }
+
+  WriteBufferMode mode;
+  std::deque<string> pending;
+  size_t totalBytes;
+  size_t writeOffset;  // Offset into the front chunk for partial writes
+};
+}  // namespace et
+
+#endif  // __ET_WRITE_BUFFER__