Preserve Windows terminal output through teardown

Author: epsalmond

CMakeLists.txt

@@ -674,13 +674,23 @@ if(WIN32)
       ${CMAKE_CURRENT_SOURCE_DIR}/test/integration_tests/*Test.cpp
       ${CMAKE_CURRENT_SOURCE_DIR}/test/unit_tests/*Test.cpp)
 
+    add_executable(
+      et-conpty-output-helper
+      test/helpers/ConPtyOutputHelper.cpp)
+
     # Same test set as Unix (see the else(WIN32) branch below): every test is
     # expected to compile and pass on Windows. Platform-impossible cases
     # SKIP at runtime with a reason instead of being excluded here.
     add_executable(et-test ${WINDOWS_TEST_SRCS} test/Main.cpp)
     target_include_directories(
       et-test PRIVATE test test/integration_tests test/unit_tests)
-    add_dependencies(et-test generated-code TerminalCommon et-lib HtmCommon)
+    add_dependencies(
+      et-test
+      generated-code
+      TerminalCommon
+      et-lib
+      HtmCommon
+      et-conpty-output-helper)
     target_link_libraries(
       et-test
       PRIVATE

src/terminal/PseudoUserTerminalWindows.hpp

@@ -221,7 +221,6 @@ class PseudoUserTerminal : public UserTerminal {
       char bytes[16 * 1024];
       DWORD count = 0;
       HANDLE output = static_cast<HANDLE>(outputRead);
-      constexpr size_t maxPendingOutput = 256 * 1024;
       // Exits only when ReadFile fails, i.e. once ClosePseudoConsole has
       // released the output pipe (see cleanup()). The pending-output cap
       // below is bypassed once closing is set, so a stalled consumer can
@@ -230,21 +229,20 @@ class PseudoUserTerminal : public UserTerminal {
       while (true) {
         {
           std::unique_lock<std::mutex> guard(pendingMutex);
-          pendingDrained.wait(guard, [this, maxPendingOutput]() {
-            return closing || pendingOutput.size() < maxPendingOutput;
+          pendingDrained.wait(guard, [this]() {
+            return closing || pendingOutput.size() < MAX_PENDING_OUTPUT;
           });
         }
         if (!ReadFile(output, bytes, sizeof(bytes), &count, NULL) ||
             count == 0) {
           break;
         }
         std::lock_guard<std::mutex> guard(pendingMutex);
-        if (closing) {
-          // Nobody will call drainOutput() again; keep reading so the host's
-          // writes never block on a full pipe, but drop earlier bytes so
-          // this can't grow unbounded while we wait for it to exit.
-          pendingOutput.clear();
-        }
+        // cleanup() terminates the process tree before closing the
+        // pseudoconsole, so bypassing the normal cap during teardown has a
+        // bounded producer. Preserve everything already queued while this
+        // thread drains the output pipe; clearing it here loses terminal
+        // output that the router may still deliver after reconnect.
         pendingOutput.append(bytes, count);
       }
     });
@@ -256,15 +254,9 @@ class PseudoUserTerminal : public UserTerminal {
   virtual void runTerminal() override {}
 
   virtual void cleanup() override {
-    // Stop enforcing the pending-output cap and wake the drain thread in
-    // case it is currently parked there. Without this, a stalled consumer
-    // (router backpressure, or nobody left to call drainOutput()) could
-    // leave the ConPTY output pipe undrained while the host process below
-    // still has queued output to write; ClosePseudoConsole then blocks
-    // waiting for a host that is itself blocked on a full pipe, hanging
-    // forever.
-    closing = true;
-    pendingDrained.notify_all();
+    // Interrupt input and the process tree before lifting the output cap. This
+    // bounds how much final output teardown can append while preserving every
+    // byte that was already queued for the router.
     if (inputWrite != INVALID_HANDLE_VALUE) {
       CloseHandle(static_cast<HANDLE>(inputWrite));
       inputWrite = INVALID_HANDLE_VALUE;
@@ -284,6 +276,12 @@ class PseudoUserTerminal : public UserTerminal {
       CloseHandle(static_cast<HANDLE>(jobHandle));
       jobHandle = nullptr;
     }
+    // Wake the reader if router backpressure parked it at the cap. It must
+    // continue draining while ClosePseudoConsole runs: older Windows waits
+    // for the pseudoconsole host, which can itself be blocked writing final
+    // output to a full pipe.
+    closing = true;
+    pendingDrained.notify_all();
     if (hPC != nullptr) {
       // Blocks until the ConPTY host process exits. The drain thread above
       // keeps calling ReadFile the whole time (closing bypasses its cap), so
@@ -368,6 +366,7 @@ class PseudoUserTerminal : public UserTerminal {
   }
 
  protected:
+  static constexpr size_t MAX_PENDING_OUTPUT = 256 * 1024;
   void* hPC;
   void* inputWrite;
   void* outputRead;

test/helpers/ConPtyOutputHelper.cpp

@@ -0,0 +1,38 @@
+#include <windows.h>
+
+#include <array>
+#include <cstddef>
+
+int main() {
+  // Open the attached console explicitly. Test runners may give a child
+  // inherited standard handles of their own, while CONOUT$ always targets the
+  // pseudoconsole selected through PROC_THREAD_ATTRIBUTE_PSEUDOCONSOLE.
+  HANDLE output =
+      CreateFileW(L"CONOUT$", GENERIC_WRITE, FILE_SHARE_READ | FILE_SHARE_WRITE,
+                  nullptr, OPEN_EXISTING, 0, nullptr);
+  if (output == nullptr || output == INVALID_HANDLE_VALUE) {
+    return 1;
+  }
+
+  std::array<char, 16 * 1024> chunk;
+  chunk.fill('A');
+  constexpr size_t totalBytes = 1024 * 1024;
+  size_t totalWritten = 0;
+  while (totalWritten < totalBytes) {
+    size_t chunkOffset = 0;
+    while (chunkOffset < chunk.size()) {
+      DWORD written = 0;
+      if (!WriteFile(output, chunk.data() + chunkOffset,
+                     static_cast<DWORD>(chunk.size() - chunkOffset), &written,
+                     nullptr) ||
+          written == 0) {
+        CloseHandle(output);
+        return 2;
+      }
+      chunkOffset += written;
+      totalWritten += written;
+    }
+  }
+  CloseHandle(output);
+  return 0;
+}

test/integration_tests/TerminalTest.cpp

@@ -706,6 +706,12 @@ class PartialFailurePipeSocketHandler : public PipeSocketHandler {
     outputToFail = output;
     partialFd = -1;
     partialWriteDone = false;
+    targetWriteFailed = false;
+  }
+
+  bool didFailTargetWrite() {
+    lock_guard<mutex> guard(failureMutex);
+    return targetWriteFailed;
   }
 
   ssize_t write(int fd, const void* buf, size_t count) override {
@@ -724,6 +730,8 @@ class PartialFailurePipeSocketHandler : public PipeSocketHandler {
         sendPartial = true;
       } else if (partialWriteDone && fd == partialFd) {
         outputToFail.clear();
+        partialFd = -1;
+        targetWriteFailed = true;
         failWrite = true;
       }
     }
@@ -744,6 +752,7 @@ class PartialFailurePipeSocketHandler : public PipeSocketHandler {
   string outputToFail;
   int partialFd = -1;
   bool partialWriteDone = false;
+  bool targetWriteFailed = false;
 };
 #endif
 
@@ -906,14 +915,16 @@ TEST_CASE_METHOD(EndToEndTestFixture, "WindowsOutputSurvivesFailedWrite",
          std::chrono::steady_clock::now() < terminalDeadline) {
     std::this_thread::sleep_for(std::chrono::milliseconds(10));
   }
+  const bool terminalWasSetup = fakeUserTerminal->isSetup();
 
   const string marker = "ET_WINDOWS_PENDING_OUTPUT";
-  if (fakeUserTerminal->isSetup()) {
+  if (terminalWasSetup) {
     routerSocketHandler->failNextWriteContaining(marker);
     fakeUserTerminal->simulateTerminalResponse(marker);
   }
 
   string received;
+  bool terminalSurvivedReconnect = false;
   const auto outputDeadline =
       std::chrono::steady_clock::now() + std::chrono::seconds(30);
   while (fakeUserTerminal->isSetup() &&
@@ -922,6 +933,8 @@ TEST_CASE_METHOD(EndToEndTestFixture, "WindowsOutputSurvivesFailedWrite",
       received += fakeConsole->getTerminalData(1);
     }
     if (received == marker) {
+      terminalSurvivedReconnect =
+          fakeUserTerminal->isSetup() && !fakeUserTerminal->wasCleanedUp();
       break;
     }
     std::this_thread::sleep_for(std::chrono::milliseconds(10));
@@ -935,8 +948,12 @@ TEST_CASE_METHOD(EndToEndTestFixture, "WindowsOutputSurvivesFailedWrite",
   handlerThread.join();
   handler.reset();
 
-  REQUIRE(fakeUserTerminal->isSetup());
+  REQUIRE(terminalWasSetup);
+  REQUIRE(routerSocketHandler->didFailTargetWrite());
   REQUIRE(received == marker);
+  REQUIRE(terminalSurvivedReconnect);
+  REQUIRE(fakeUserTerminal->wasCleanedUp());
+  REQUIRE_FALSE(fakeUserTerminal->isSetup());
 }
 #endif
 

test/unit_tests/PseudoUserTerminalWindowsTest.cpp

@@ -2,6 +2,7 @@
 
 #include <windows.h>
 
+#include <array>
 #include <chrono>
 #include <future>
 #include <thread>
@@ -13,10 +14,12 @@ namespace et {
 
 class OutputPressureTerminal : public PseudoUserTerminal {
  public:
-  size_t queuedOutputSize() {
+  std::string queuedOutput() {
     std::lock_guard<std::mutex> guard(pendingMutex);
-    return pendingOutput.size();
+    return pendingOutput;
   }
+
+  static constexpr size_t outputLimit() { return MAX_PENDING_OUTPUT; }
 };
 
 TEST_CASE("PseudoUserTerminal cleanup falls back without a job",
@@ -53,35 +56,47 @@ TEST_CASE("PseudoUserTerminal cleanup falls back without a job",
 // parked waiting for drainOutput() to free up its pending-output cap, and
 // nobody calls drainOutput() again during shutdown), the ConPTY host can
 // block writing to a full pipe and ClosePseudoConsole waits for that host
-// to exit forever. Heap-allocate the terminal and leak it (rather than
-// join/delete) if cleanup() does not return in time, since a hung cleanup()
-// thread could otherwise touch freed memory.
+// to exit forever. Teardown must continue draining without discarding output
+// that was already queued for the router. Heap-allocate the terminal and leak
+// it (rather than join/delete) if cleanup() does not return in time, since a
+// hung cleanup() thread could otherwise touch freed memory.
 TEST_CASE(
     "PseudoUserTerminal cleanup does not hang under queued, undrained output",
     "[PseudoUserTerminal][windows]") {
   const char* previousShell = ::getenv("SHELL");
   const bool hadShell = previousShell != nullptr;
   const std::string savedShell = hadShell ? previousShell : "";
-  REQUIRE(_putenv_s("SHELL", "cmd.exe") == 0);
+  std::array<wchar_t, 32768> testExecutable;
+  const DWORD testExecutableLength =
+      GetModuleFileNameW(nullptr, testExecutable.data(),
+                         static_cast<DWORD>(testExecutable.size()));
+  REQUIRE(testExecutableLength > 0);
+  REQUIRE(testExecutableLength < testExecutable.size());
+  const fs::path helperPath =
+      fs::path(std::wstring(testExecutable.data(), testExecutableLength))
+          .parent_path() /
+      "et-conpty-output-helper.exe";
+  REQUIRE(fs::is_regular_file(helperPath));
+  REQUIRE(fs::is_directory(helperPath.parent_path()));
+  REQUIRE(_putenv_s("SHELL", helperPath.string().c_str()) == 0);
   auto* term = new OutputPressureTerminal();
   const int setupResult = term->setup(-1);
   REQUIRE(_putenv_s("SHELL", savedShell.c_str()) == 0);
   REQUIRE(setupResult == 0);
 
-  // Flood stdout continuously and never call drainOutput(), modeling a
-  // stalled router consumer. This drives the drain thread past its
-  // pending-output cap and fills the underlying pipe before teardown.
-  term->writeInput(
-      "for /L %i in (1,1,1000000) do @echo "
-      "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
-      "\r\n");
+  // The helper writes a finite 1 MiB directly to its ConPTY stdout. Never
+  // call drainOutput() before cleanup, modeling a stalled router consumer
+  // without relying on shell startup, command parsing, or loop throughput.
   const auto pressureDeadline =
-      std::chrono::steady_clock::now() + std::chrono::seconds(5);
-  while (term->queuedOutputSize() < 256 * 1024 &&
+      std::chrono::steady_clock::now() + std::chrono::seconds(15);
+  string queuedBeforeCleanup;
+  while (queuedBeforeCleanup.size() < OutputPressureTerminal::outputLimit() &&
          std::chrono::steady_clock::now() < pressureDeadline) {
     std::this_thread::sleep_for(std::chrono::milliseconds(20));
+    queuedBeforeCleanup = term->queuedOutput();
   }
-  const bool capReached = term->queuedOutputSize() >= 256 * 1024;
+  const bool capReached =
+      queuedBeforeCleanup.size() >= OutputPressureTerminal::outputLimit();
 
   auto done = std::make_shared<std::promise<void>>();
   std::future<void> doneFuture = done->get_future();
@@ -91,14 +106,22 @@ TEST_CASE(
   });
 
   const auto status = doneFuture.wait_for(std::chrono::seconds(15));
+  bool queuedOutputPreserved = false;
   if (status == std::future_status::ready) {
     cleanupThread.join();
+    const string queuedAfterCleanup = term->drainOutput();
+    queuedOutputPreserved =
+        queuedAfterCleanup.size() >= queuedBeforeCleanup.size() &&
+        std::equal(queuedBeforeCleanup.begin(), queuedBeforeCleanup.end(),
+                   queuedAfterCleanup.begin());
     delete term;
   } else {
     cleanupThread.detach();
   }
+  INFO("queued before cleanup: " << queuedBeforeCleanup.size());
   REQUIRE(capReached);
   REQUIRE(status == std::future_status::ready);
+  REQUIRE(queuedOutputPreserved);
 }
 
 }  // namespace et