grpc ensure FIFO order of jobs stored by the server

cpp/src/grpc/server/grpc_job_management.cpp

@@ -64,6 +64,19 @@ bool recv_incumbent_pipe(int fd, std::vector<uint8_t>& data)
 // Job management
 // =============================================================================
 
+// Assign the next ticket and publish the slot. Call only after the payload and
+// tracker entry are in place. The lock is held across both steps so ticket N+1
+// cannot become visible while ticket N is still unpublished.
+static uint64_t publish_ready_job_slot(JobQueueEntry& job)
+{
+  std::lock_guard<std::mutex> lock(job_enqueue_mutex);
+  const uint64_t seq = next_job_enqueue_seq.fetch_add(1, std::memory_order_relaxed);
+  job.enqueue_seq.store(seq, std::memory_order_relaxed);
+  job.ready.store(true, std::memory_order_release);
+  job.claimed.store(false, std::memory_order_release);
+  return seq;
+}
+
 // Reserve a shared-memory job queue slot, store the serialized request data,
 // and register the job in the tracker. Returns {true, job_id} on success.
 // Uses CAS on `claimed` for lock-free slot reservation and release semantics
@@ -114,11 +127,13 @@ std::pair<bool, std::string> submit_job_async(std::vector<uint8_t>&& request_dat
     job_tracker[job_id]   = std::move(info);
   }
 
-  // Publish: release makes all writes above visible to the dispatch thread.
-  job_queue[slot].ready.store(true, std::memory_order_release);
-  job_queue[slot].claimed.store(false, std::memory_order_release);
+  const uint64_t seq = publish_ready_job_slot(job_queue[slot]);
 
-  if (config.verbose) { SERVER_LOG_DEBUG("[Server] Job submitted (async): %s", job_id.c_str()); }
+  if (config.verbose) {
+    SERVER_LOG_DEBUG("[Server] Job submitted (async): %s enqueue_seq=%lu",
+                     job_id.c_str(),
+                     static_cast<unsigned long>(seq));
+  }
 
   return {true, job_id};
 }
@@ -170,11 +185,12 @@ std::pair<bool, std::string> submit_chunked_job_async(PendingChunkedUpload&& chu
     job_tracker[job_id]   = std::move(info);
   }
 
-  job_queue[slot].ready.store(true, std::memory_order_release);
-  job_queue[slot].claimed.store(false, std::memory_order_release);
+  const uint64_t seq = publish_ready_job_slot(job_queue[slot]);
 
   if (config.verbose) {
-    SERVER_LOG_DEBUG("[Server] Chunked job submitted (async): %s", job_id.c_str());
+    SERVER_LOG_DEBUG("[Server] Chunked job submitted (async): %s enqueue_seq=%lu",
+                     job_id.c_str(),
+                     static_cast<unsigned long>(seq));
   }
 
   return {true, job_id};

cpp/src/grpc/server/grpc_job_queue.hpp

@@ -0,0 +1,96 @@
+/*
+ * SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+#pragma once
+
+#ifdef CUOPT_ENABLE_GRPC
+
+#include <sys/types.h>
+#include <atomic>
+#include <cstddef>
+#include <cstdint>
+#include <limits>
+
+// Shared-memory job slot used by the gRPC parent and workers.
+// Slot index is ownership only; dispatch order uses enqueue_seq.
+struct JobQueueEntry {
+  char job_id[64];
+  uint32_t problem_category;      // cuopt::remote::ProblemCategory (LP, MIP, ...)
+  uint64_t data_size;             // Size of problem data (uint64 for large problems)
+  std::atomic<bool> ready;        // Job is ready to be processed
+  std::atomic<bool> claimed;      // Worker has claimed this job
+  std::atomic<pid_t> worker_pid;  // PID of worker that claimed this job (0 if none)
+  std::atomic<bool> cancelled;    // Job has been cancelled (worker should skip)
+  std::atomic<int> worker_index;  // Index of worker that claimed this job (-1 if none)
+  std::atomic<bool> data_sent;    // Server has sent data to worker's pipe
+  std::atomic<bool> is_chunked;   // True when data is in pending_chunked_data (streamed to pipe)
+  std::atomic<bool>
+    unified_header;  // Chunked pipe header is SubmitJobRequest, not ChunkedProblemHeader
+  // Monotonic submission order. 0 means the slot is free or has been reset;
+  // workers never dispatch a ready entry with enqueue_seq == 0.
+  std::atomic<uint64_t> enqueue_seq;
+};
+
+// Reset every field in a job slot so it can be reused by the next submission.
+// Clear ready before enqueue_seq so a concurrent claim cannot observe a
+// recycled slot as still dispatchable under the old sequence.
+inline void reset_job_slot(JobQueueEntry& job)
+{
+  job.ready.store(false, std::memory_order_release);
+  job.enqueue_seq.store(0, std::memory_order_relaxed);
+  job.worker_pid     = 0;
+  job.worker_index   = -1;
+  job.data_sent      = false;
+  job.is_chunked     = false;
+  job.unified_header = false;
+  job.claimed        = false;
+  job.cancelled      = false;
+}
+
+// Claim the ready, unclaimed job with the lowest enqueue_seq.
+// Retries when a concurrent claim or slot recycle invalidates the selection.
+// Returns the slot index, or -1 when no dispatchable job is present.
+inline int claim_ready_job_slot(JobQueueEntry* queue,
+                                size_t n_jobs,
+                                int worker_id,
+                                pid_t worker_pid)
+{
+  while (true) {
+    int best_slot     = -1;
+    uint64_t best_seq = std::numeric_limits<uint64_t>::max();
+
+    for (size_t i = 0; i < n_jobs; ++i) {
+      // Acquire on ready publishes the enqueue_seq written before ready=true.
+      if (!queue[i].ready.load(std::memory_order_acquire)) { continue; }
+      if (queue[i].claimed.load(std::memory_order_acquire)) { continue; }
+      const uint64_t seq = queue[i].enqueue_seq.load(std::memory_order_relaxed);
+      if (seq == 0) { continue; }
+      if (best_slot < 0 || seq < best_seq) {
+        best_seq  = seq;
+        best_slot = static_cast<int>(i);
+      }
+    }
+
+    if (best_slot < 0) { return -1; }
+
+    bool expected = false;
+    if (!queue[best_slot].claimed.compare_exchange_strong(
+          expected, true, std::memory_order_acq_rel)) {
+      continue;
+    }
+
+    if (!queue[best_slot].ready.load(std::memory_order_acquire) ||
+        queue[best_slot].enqueue_seq.load(std::memory_order_relaxed) != best_seq) {
+      queue[best_slot].claimed.store(false, std::memory_order_release);
+      continue;
+    }
+
+    queue[best_slot].worker_pid.store(worker_pid, std::memory_order_relaxed);
+    queue[best_slot].worker_index.store(worker_id, std::memory_order_relaxed);
+    return best_slot;
+  }
+}
+
+#endif  // CUOPT_ENABLE_GRPC

cpp/src/grpc/server/grpc_server_main.cpp

@@ -233,6 +233,7 @@ int main(int argc, char** argv)
       job_queue[i].claimed.store(false);
       job_queue[i].cancelled.store(false);
       job_queue[i].worker_index.store(-1);
+      job_queue[i].enqueue_seq.store(0);
     }
 
     for (size_t i = 0; i < MAX_RESULTS; ++i) {

cpp/src/grpc/server/grpc_server_types.hpp

@@ -17,6 +17,7 @@
 #include <cuopt/mathematical_optimization/optimization_problem_interface.hpp>
 #include <cuopt/mathematical_optimization/solve.hpp>
 #include <cuopt/mathematical_optimization/utilities/internals.hpp>
+#include "grpc_job_queue.hpp"
 #include "grpc_problem_mapper.hpp"
 #include "grpc_server_logger.hpp"
 #include "grpc_settings_mapper.hpp"
@@ -83,21 +84,6 @@ void copy_cstr(char (&dst)[N], const char* src)
   std::snprintf(dst, N, "%s", src ? src : "");
 }
 
-struct JobQueueEntry {
-  char job_id[64];
-  uint32_t problem_category;      // cuopt::remote::ProblemCategory (LP, MIP, ...)
-  uint64_t data_size;             // Size of problem data (uint64 for large problems)
-  std::atomic<bool> ready;        // Job is ready to be processed
-  std::atomic<bool> claimed;      // Worker has claimed this job
-  std::atomic<pid_t> worker_pid;  // PID of worker that claimed this job (0 if none)
-  std::atomic<bool> cancelled;    // Job has been cancelled (worker should skip)
-  std::atomic<int> worker_index;  // Index of worker that claimed this job (-1 if none)
-  std::atomic<bool> data_sent;    // Server has sent data to worker's pipe
-  std::atomic<bool> is_chunked;   // True when data is in pending_chunked_data (streamed to pipe)
-  std::atomic<bool>
-    unified_header;  // Chunked pipe header is SubmitJobRequest, not ChunkedProblemHeader
-};
-
 enum ResultStatus : uint32_t {
   RESULT_SUCCESS   = 0,
   RESULT_ERROR     = 1,
@@ -253,6 +239,12 @@ inline std::mutex waiters_mutex;
 inline JobQueueEntry* job_queue       = nullptr;
 inline ResultQueueEntry* result_queue = nullptr;
 inline SharedMemoryControl* shm_ctrl  = nullptr;
+// Parent-process only: assigns enqueue_seq when accepting a job. Starts at 1 so
+// 0 remains the sentinel for free/reset slots. job_enqueue_mutex is held from
+// the ticket assignment until ready is published, so a later ticket cannot
+// become visible before an earlier one.
+inline std::atomic<uint64_t> next_job_enqueue_seq{1};
+inline std::mutex job_enqueue_mutex;
 // Per-slot ready flags: child sets after CUDA/RMM init and the startup GPU
 // probe. Parent clears before a respawn and waits for the new child to publish.
 // Sized to config.num_workers; do not use active_workers (SIGKILL leaves it stale).

cpp/src/grpc/server/grpc_worker.cpp

@@ -318,19 +318,6 @@ class IncumbentPipeCallback : public cuopt::internals::get_solution_callback_t {
 // Small utility helpers
 // ---------------------------------------------------------------------------
 
-// Reset every field in a job slot so it can be reused by the next submission.
-static void reset_job_slot(JobQueueEntry& job)
-{
-  job.worker_pid     = 0;
-  job.worker_index   = -1;
-  job.data_sent      = false;
-  job.is_chunked     = false;
-  job.unified_header = false;
-  job.ready          = false;
-  job.claimed        = false;
-  job.cancelled      = false;
-}
-
 // Log pipe throughput when config.verbose is enabled.
 static void log_pipe_throughput(const char* phase,
                                 int64_t total_bytes,
@@ -412,21 +399,10 @@ static void store_simple_result(const std::string& job_id,
 // Stage functions called from the worker_process main loop
 // ---------------------------------------------------------------------------
 
-// Atomically claim the first ready-but-unclaimed job slot, stamping it with
-// this worker's PID and index.  Returns the slot index, or -1 if none found.
+// Atomically claim the oldest ready-but-unclaimed job by enqueue_seq.
 static int claim_job_slot(int worker_id)
 {
-  for (size_t i = 0; i < MAX_JOBS; ++i) {
-    if (job_queue[i].ready && !job_queue[i].claimed) {
-      bool expected = false;
-      if (job_queue[i].claimed.compare_exchange_strong(expected, true)) {
-        job_queue[i].worker_pid   = getpid();
-        job_queue[i].worker_index = worker_id;
-        return static_cast<int>(i);
-      }
-    }
-  }
-  return -1;
+  return claim_ready_job_slot(job_queue, MAX_JOBS, worker_id, getpid());
 }
 
 // Unified chunked upload: the pipe header is a SubmitJobRequest with numeric

cpp/src/grpc/server/grpc_worker_infra.cpp

@@ -686,14 +686,7 @@ void mark_worker_jobs_failed(pid_t dead_worker_pid)
       }
 
       // 3. Release the job queue slot and update the in-process job tracker.
-      job_queue[i].worker_pid     = 0;
-      job_queue[i].worker_index   = -1;
-      job_queue[i].data_sent      = false;
-      job_queue[i].is_chunked     = false;
-      job_queue[i].unified_header = false;
-      job_queue[i].ready          = false;
-      job_queue[i].claimed        = false;
-      job_queue[i].cancelled      = false;
+      reset_job_slot(job_queue[i]);
 
       {
         std::lock_guard<std::mutex> lock(tracker_mutex);

cpp/tests/linear_programming/grpc/CMakeLists.txt

@@ -127,6 +127,32 @@ install(
     EXCLUDE_FROM_ALL
 )
 
+# ##################################################################################################
+# - gRPC job queue ordering ------------------------------------------------------------------------
+# Header-only checks of claim_ready_job_slot; no server or GPU is required.
+
+add_executable(GRPC_JOB_QUEUE_TEST ${CMAKE_CURRENT_SOURCE_DIR}/grpc_job_queue_test.cpp)
+
+target_compile_definitions(GRPC_JOB_QUEUE_TEST PRIVATE CUOPT_ENABLE_GRPC)
+
+target_include_directories(GRPC_JOB_QUEUE_TEST PRIVATE "${CUOPT_SOURCE_DIR}/src/grpc/server")
+
+target_link_libraries(GRPC_JOB_QUEUE_TEST PRIVATE GTest::gtest GTest::gtest_main Threads::Threads)
+
+if(NOT DEFINED INSTALL_TARGET OR "${INSTALL_TARGET}" STREQUAL "")
+  target_link_options(GRPC_JOB_QUEUE_TEST PRIVATE -Wl,--enable-new-dtags)
+endif()
+
+rapids_test_add(NAME GRPC_JOB_QUEUE_TEST COMMAND GRPC_JOB_QUEUE_TEST INSTALL_COMPONENT_SET testing)
+set_tests_properties(GRPC_JOB_QUEUE_TEST PROPERTIES LABELS "numopt")
+
+install(
+    TARGETS GRPC_JOB_QUEUE_TEST
+    COMPONENT testing
+    DESTINATION bin/gtests/libcuopt
+    EXCLUDE_FROM_ALL
+)
+
 # ##################################################################################################
 # - gRPC Integration Tests -------------------------------------------------------------------------
 # These tests start a real server process and run end-to-end tests