test: clean up gRPC server process groups

Signed-off-by: Miles Lubin <mlubin@nvidia.com>

Author: mlubin

cpp/tests/linear_programming/grpc/grpc_integration_test.cpp

@@ -29,6 +29,7 @@
 #include <gtest/gtest.h>
 
 #include <atomic>
+#include <cerrno>
 #include <cmath>
 #include <filesystem>
 #include <fstream>
@@ -71,14 +72,22 @@ using cuopt::mathematical_optimization::testing::GrpcTestLogCapture;
 
 namespace {
 
-// =============================================================================
-// Server Process Manager
-// =============================================================================
+// GRPC_INTEGRATION_TEST
+//   `-- cuopt_grpc_server parent (pid_, leads process group pgid_ == pid_)
+//         `-- GPU worker (inherits pgid_)
+//
+// The server is forked into its own process group so stop() can signal the whole
+// group and guarantee the GPU worker child is never left orphaned.
 
 class ServerProcess {
  public:
-  ServerProcess() : pid_(-1), port_(0) {}
-  ~ServerProcess() { stop(); }
+  ServerProcess() : pid_(-1), pgid_(-1), port_(0) {}
+  ServerProcess(const ServerProcess&)            = delete;
+  ServerProcess& operator=(const ServerProcess&) = delete;
+  ~ServerProcess()
+  {
+    if (!stop()) { std::cerr << "Failed to clean up test server process group\n"; }
+  }
 
   void set_tls_config(const std::string& root_certs,
                       const std::string& client_cert = "",
@@ -91,21 +100,27 @@ class ServerProcess {
 
   bool start(int port, const std::vector<std::string>& extra_args = {})
   {
-    port_ = port;
+    if (pid_ > 0) {
+      std::cerr << "Cannot reuse a ServerProcess while it still owns a process lifecycle\n";
+      return false;
+    }
 
     std::string server_path = find_server_binary();
     if (server_path.empty()) {
       std::cerr << "Could not find cuopt_grpc_server binary\n";
       return false;
     }
 
-    pid_ = fork();
+    port_ = port;
+    pid_  = fork();
     if (pid_ < 0) {
       std::cerr << "fork() failed\n";
       return false;
     }
 
     if (pid_ == 0) {
+      setpgid(0, 0);  // child leads its own group; parent mirrors this below
+
       std::vector<const char*> args;
       args.push_back(server_path.c_str());
       args.push_back("--port");
@@ -131,33 +146,42 @@ class ServerProcess {
       _exit(127);
     }
 
-    return wait_for_ready(15000);
+    // Mirror the child's setpgid() so the group is established regardless of which
+    // process runs first; the child is a fresh fork, so its group id is its pid.
+    setpgid(pid_, pid_);
+    pgid_ = pid_;
+
+    if (!wait_for_ready(15000)) {
+      if (!stop()) { std::cerr << "Failed to clean up server after readiness failure\n"; }
+      return false;
+    }
+
+    return true;
   }
 
-  void stop()
+  bool stop()
   {
-    if (pid_ > 0) {
-      kill(pid_, SIGTERM);
-
-      int status;
-      int wait_ms = 0;
-      while (wait_ms < 5000) {
-        int ret = waitpid(pid_, &status, WNOHANG);
-        if (ret != 0) break;
-        std::this_thread::sleep_for(std::chrono::milliseconds(100));
-        wait_ms += 100;
-      }
+    if (pid_ <= 0) return true;
+
+    // Ask the whole group (server + GPU worker) to exit, escalating to SIGKILL if
+    // the server does not honor SIGTERM within the timeout.
+    kill(-pgid_, SIGTERM);
+    if (!reap_parent(std::chrono::seconds(5))) {
+      kill(-pgid_, SIGKILL);
+      int status = 0;
+      while (waitpid(pid_, &status, 0) < 0 && errno == EINTR) {}
+    }
 
-      if (waitpid(pid_, &status, WNOHANG) == 0) {
-        kill(pid_, SIGKILL);
-        waitpid(pid_, &status, 0);
-      }
+    // With the server reaped, the group is gone once the GPU worker exits too.
+    bool group_exited = wait_for_group_exit(std::chrono::seconds(5));
+    if (!group_exited) { std::cerr << "Server process group " << pgid_ << " did not exit\n"; }
 
-      pid_ = -1;
-    }
+    clear_lifecycle_state();
+    return group_exited;
   }
 
   int port() const { return port_; }
+  pid_t verified_process_group_id() const { return pgid_; }
 
   bool is_running() const
   {
@@ -172,6 +196,40 @@ class ServerProcess {
   }
 
  private:
+  void clear_lifecycle_state()
+  {
+    pid_  = -1;
+    pgid_ = -1;
+    port_ = 0;
+  }
+
+  // Reap the server parent, returning true once it has exited (within timeout).
+  bool reap_parent(std::chrono::milliseconds timeout)
+  {
+    auto deadline = std::chrono::steady_clock::now() + timeout;
+    while (true) {
+      int status = 0;
+      pid_t ret  = waitpid(pid_, &status, WNOHANG);
+      if (ret == pid_) return true;
+      if (ret < 0 && errno == ECHILD) return true;  // already reaped elsewhere
+      if (ret < 0 && errno == EINTR) continue;
+      if (std::chrono::steady_clock::now() >= deadline) return false;
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+  }
+
+  // True once no process remains in the group. The parent must already be reaped,
+  // otherwise its zombie keeps the group id alive.
+  bool wait_for_group_exit(std::chrono::milliseconds timeout) const
+  {
+    auto deadline = std::chrono::steady_clock::now() + timeout;
+    while (true) {
+      if (kill(-pgid_, 0) != 0 && errno == ESRCH) return true;
+      if (std::chrono::steady_clock::now() >= deadline) return false;
+      std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+  }
+
   std::string find_in_path(const std::string& name)
   {
     const char* path_env = std::getenv("PATH");
@@ -246,8 +304,8 @@ class ServerProcess {
       grpc_client_t client(config);
       if (client.connect()) { return true; }
 
-      int status;
-      if (waitpid(pid_, &status, WNOHANG) != 0) {
+      int status = 0;
+      if (waitpid(pid_, &status, WNOHANG) == pid_) {
         std::cerr << "Server process died during startup\n";
         return false;
       }
@@ -257,6 +315,7 @@ class ServerProcess {
   }
 
   pid_t pid_;
+  pid_t pgid_;
   int port_;
   std::string tls_root_certs_;
   std::string tls_client_cert_;
@@ -516,7 +575,7 @@ class DefaultServerTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }
 
@@ -1083,7 +1142,7 @@ class ChunkedUploadTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }
 
@@ -1399,7 +1458,7 @@ class PathSelectionTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }
 
@@ -1576,14 +1635,51 @@ TEST_F(PathSelectionTests, UnaryUploadMIPWithPathLogging)
     << log_capture.get_server_logs();
 }
 
+// =============================================================================
+// Server Process Lifecycle Tests
+// =============================================================================
+
+class ServerProcessLifecycleTests : public GrpcIntegrationTestBase {
+ protected:
+  void SetUp() override { port_ = get_test_port(); }
+  void TearDown() override { EXPECT_TRUE(server_.stop()); }
+
+  ServerProcess server_;
+};
+
+::testing::AssertionResult VerifiedProcessGroupIsGone(pid_t pgid)
+{
+  errno = 0;
+  if (kill(-pgid, 0) < 0 && errno == ESRCH) { return ::testing::AssertionSuccess(); }
+  return ::testing::AssertionFailure()
+         << "Verified server process group " << pgid << " still exists (errno " << errno << ")";
+}
+
+TEST_F(ServerProcessLifecycleTests, IdleServerStopCleansProcessGroup)
+{
+  ASSERT_TRUE(server_.start(port_));
+  pid_t server_pgid = server_.verified_process_group_id();
+  ASSERT_GT(server_pgid, 0);
+  EXPECT_TRUE(server_.stop());
+  EXPECT_FALSE(server_.is_running());
+  EXPECT_TRUE(VerifiedProcessGroupIsGone(server_pgid));
+}
+
+TEST_F(ServerProcessLifecycleTests, StopIsIdempotent)
+{
+  ASSERT_TRUE(server_.start(port_));
+  ASSERT_TRUE(server_.stop());
+  EXPECT_TRUE(server_.stop());
+}
+
 // =============================================================================
 // Error Recovery Tests (per-test server lifecycle)
 // =============================================================================
 
 class ErrorRecoveryTests : public GrpcIntegrationTestBase {
  protected:
   void SetUp() override { port_ = get_test_port(); }
-  void TearDown() override { server_.stop(); }
+  void TearDown() override { EXPECT_TRUE(server_.stop()); }
 
   bool start_server(const std::vector<std::string>& extra_args = {})
   {
@@ -1602,7 +1698,7 @@ TEST_F(ErrorRecoveryTests, ClientReconnectsAfterServerRestart)
   auto status_before = client->check_status("test-job");
   EXPECT_TRUE(status_before.success);
 
-  server_.stop();
+  ASSERT_TRUE(server_.stop());
   EXPECT_FALSE(server_.is_running());
 
   auto status_down = client->check_status("test-job");
@@ -1617,6 +1713,8 @@ TEST_F(ErrorRecoveryTests, ClientReconnectsAfterServerRestart)
 TEST_F(ErrorRecoveryTests, ClientHandlesServerCrashDuringSolve)
 {
   ASSERT_TRUE(start_server());
+  pid_t server_pgid = server_.verified_process_group_id();
+  ASSERT_GT(server_pgid, 0);
   auto client = create_client();
   ASSERT_NE(client, nullptr);
 
@@ -1630,7 +1728,8 @@ TEST_F(ErrorRecoveryTests, ClientHandlesServerCrashDuringSolve)
   ASSERT_TRUE(submit_result.success);
 
   std::this_thread::sleep_for(std::chrono::milliseconds(500));
-  server_.stop();
+  EXPECT_TRUE(server_.stop()) << "Active-solve shutdown must remove the server process group";
+  EXPECT_TRUE(VerifiedProcessGroupIsGone(server_pgid));
 
   auto status_result = client->check_status(submit_result.job_id);
   EXPECT_FALSE(status_result.success);
@@ -1694,7 +1793,7 @@ TEST_F(ErrorRecoveryTests, ChunkedUploadAfterServerRestart)
   auto result1 = client->solve_mip(problem, settings, false);
   EXPECT_TRUE(result1.success) << result1.error_message;
 
-  server_.stop();
+  ASSERT_TRUE(server_.stop());
   ASSERT_TRUE(start_server({"--max-message-mb", "256"}));
 
   auto client2 = create_client(config);
@@ -1745,7 +1844,7 @@ class TlsServerTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }
 
@@ -1846,7 +1945,7 @@ class MtlsServerTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }
 
@@ -1917,7 +2016,7 @@ class ChunkValidationTests : public GrpcIntegrationTestBase {
 
   static void TearDownTestSuite()
   {
-    if (s_server_) s_server_->stop();
+    if (s_server_) EXPECT_TRUE(s_server_->stop());
     s_server_.reset();
   }