fix(grpc): drain all remaining log lines at job completion in StreamLogs (#1508)

## Summary

- `StreamLogs` did a single extra `getline` after detecting job completion, silently dropping any lines flushed between the last EOF poll and the job-complete marker
- With multiple final log lines (e.g. `"Optimal solution found."` followed by `"Best objective ..."`), only the first was streamed to the client
- Replace the single read with a drain loop so all buffered lines are sent before the `job_complete` sentinel

## Root cause

`test_cli_lp_remote` and `test_cli_mip_remote` both parse the CLI subprocess output for solver status and objective. In remote mode the CLI streams server logs via gRPC. The race: if the solver logs two final lines in quick succession and the job is marked complete before `StreamLogs` polls again, the old code sent exactly one of those lines and dropped the rest.

The probability scales with I/O load (more xdist workers = more contention = wider race window), which is why it surfaced when experimenting with higher `-n` values in pytest-xdist.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Authors:
  - Ramakrishna Prabhu (https://github.com/ramakrishnap-nv)

Approvers:
  - Alice Boucher (https://github.com/aliceb-nv)

URL: #1508

Author: ramakrishnap-nv

cpp/src/grpc/client/grpc_client.cpp

@@ -281,12 +281,14 @@ void grpc_client_t::start_log_streaming(const std::string& job_id)
   }
 
   stop_logs_.store(false);
+  log_stream_done_.store(false);
   log_thread_ = std::make_unique<std::thread>([this, job_id]() {
     stream_logs(job_id, 0, [this](const std::string& line, bool /*job_complete*/) {
       if (stop_logs_.load()) return false;
       if (config_.log_callback) { config_.log_callback(line); }
       return true;
     });
+    log_stream_done_.store(true);
   });
 }
 
@@ -308,6 +310,28 @@ void grpc_client_t::stop_log_streaming()
   if (t && t->joinable()) { t->join(); }
 }
 
+void grpc_client_t::drain_log_streaming()
+{
+  // No-op when streaming was never started (config_.stream_logs or
+  // config_.log_callback disabled): log_stream_done_ stays false but
+  // log_thread_ is null, so stop_log_streaming() is also a no-op.
+  // Avoid the polling loop entirely to prevent a 5-second delay on the
+  // common non-streaming solve path.
+  if (!log_thread_) return;
+
+  // On the success path the server drains all pending log lines and then sends
+  // a job_complete sentinel; the streaming thread exits naturally when that
+  // sentinel arrives.  Poll for natural completion (up to kDrainTimeout) so
+  // callers receive the final log lines (e.g. "Best objective …") before we
+  // force-cancel.  Fall back to stop_log_streaming() if the deadline passes.
+  static constexpr auto kDrainTimeout = std::chrono::seconds(5);
+  auto deadline                       = std::chrono::steady_clock::now() + kDrainTimeout;
+  while (!log_stream_done_.load() && std::chrono::steady_clock::now() < deadline) {
+    std::this_thread::sleep_for(std::chrono::milliseconds(10));
+  }
+  stop_log_streaming();
+}
+
 // =============================================================================
 // Proto → Client Enum Conversion
 // =============================================================================
@@ -848,7 +872,13 @@ remote_lp_result_t<i_t, f_t> grpc_client_t::solve_lp(
 
   start_log_streaming(sub.job_id);
   auto poll = poll_for_completion(sub.job_id);
-  stop_log_streaming();
+  // On success, wait for the server to send the job_complete sentinel so all
+  // final log lines are received before the stream is torn down.
+  if (poll.completed) {
+    drain_log_streaming();
+  } else {
+    stop_log_streaming();
+  }
 
   if (!poll.completed) { return {.error_message = poll.error_message}; }
 
@@ -875,7 +905,11 @@ remote_mip_result_t<i_t, f_t> grpc_client_t::solve_mip(
 
   start_log_streaming(sub.job_id);
   auto poll = poll_for_completion(sub.job_id);
-  stop_log_streaming();
+  if (poll.completed) {
+    drain_log_streaming();
+  } else {
+    stop_log_streaming();
+  }
 
   if (!poll.completed) { return {.error_message = poll.error_message}; }
 

cpp/src/grpc/client/grpc_client.hpp

@@ -408,6 +408,10 @@ class grpc_client_t {
   // Activated when config_.stream_logs is true and config_.log_callback is set.
   void start_log_streaming(const std::string& job_id);
   void stop_log_streaming();
+  // Graceful variant: waits up to kLogDrainTimeout for the server to send the
+  // job_complete sentinel before force-cancelling.  Use on the success path so
+  // final log lines (e.g. "Best objective …") are not dropped.
+  void drain_log_streaming();
 
   // Shared polling loop used by solve_lp and solve_mip.
   struct poll_result_t {
@@ -419,6 +423,7 @@ class grpc_client_t {
 
   std::unique_ptr<std::thread> log_thread_;
   std::atomic<bool> stop_logs_{false};
+  std::atomic<bool> log_stream_done_{false};
   mutable std::mutex log_context_mutex_;
   // Points to the grpc::ClientContext* of the in-flight StreamLogs RPC (if
   // any).  Typed as void* to avoid exposing grpc headers in the public API.

cpp/src/grpc/server/grpc_service_impl.cpp

@@ -823,31 +823,33 @@ class CuOptRemoteServiceImpl final : public cuopt::remote::CuOptRemoteService::S
         in.clear();
       }
 
-      // Job finished while we were caught up at EOF. The worker may still flush
-      // the last log line after marking the job complete, so read once more
-      // before closing the stream.
-      std::string msg;
-      JobStatus s = check_job_status(job_id, msg);
-      if (s == JobStatus::COMPLETED || s == JobStatus::FAILED || s == JobStatus::CANCELLED) {
-        // Resume offset for the final read (same tellg logic as the main loop).
+      // Job finished while we were caught up at EOF. Drain all remaining lines
+      // before closing: the worker may have flushed several lines (e.g. the
+      // final solver summary) between our last getline and the job-complete
+      // marker, and a single extra read would silently drop everything after
+      // the first of those lines.
+      std::string term_msg;
+      JobStatus term_status = check_job_status(job_id, term_msg);
+      if (term_status == JobStatus::COMPLETED || term_status == JobStatus::FAILED ||
+          term_status == JobStatus::CANCELLED) {
         std::streampos read_start = in.tellg();
         if (read_start >= 0) { current_offset = static_cast<int64_t>(read_start); }
 
-        // One last getline: picks up a trailing line written after our previous
-        // EOF poll. If nothing remains, skip straight to the sentinel below.
-        if (std::getline(in, line)) {
-          std::streampos read_end = in.tellg();
-          // byte_offset on each message is where the client should resume; mirror
-          // the main-loop fallback when tellg() is unavailable after the last line.
+        bool drain_ok = true;
+        while (std::getline(in, line)) {
+          std::streampos read_end  = in.tellg();
           int64_t next_byte_offset = current_offset + static_cast<int64_t>(line.size());
           if (read_end >= 0) { next_byte_offset = static_cast<int64_t>(read_end); }
-          (void)write_log_message(line, next_byte_offset, false);
+          if (!write_log_message(line, next_byte_offset, false)) {
+            drain_ok = false;
+            break;
+          }
+          current_offset = next_byte_offset;
         }
 
         // Empty line + job_complete=true tells the client the log stream is done.
-        // Use current_offset (not next_byte_offset): if we sent a final line above,
-        // the client already got its resume point; this sentinel only marks EOF.
-        (void)write_log_message("", current_offset);
+        // Skip sentinel if the Write() failed mid-drain — stream is already broken.
+        if (drain_ok) { (void)write_log_message("", current_offset); }
         return Status::OK;
       }