test: Add test case for RequestTracker counter mismatch

qa/L0_simple_ensemble/ensemble_backpressure_test.py

@@ -30,6 +30,7 @@
 
 sys.path.append("../common")
 
+import os
 import queue
 import threading
 import time
@@ -45,6 +46,8 @@
 SERVER_URL = "localhost:8001"
 DEFAULT_RESPONSE_TIMEOUT = 60
 EXPECTED_INFER_OUTPUT = 0.5
+MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE = "ensemble_parallel_step_failed_enqueue"
+EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT = 4.0
 
 NUM_REQUESTS = 16
 NUM_RESPONSES_PER_REQUEST = 8
@@ -76,19 +79,17 @@ def prepare_infer_args(input_value, enable_batching=False):
     return infer_input, outputs
 
 
-def collect_responses(user_data):
+def collect_responses(user_data, timeout=DEFAULT_RESPONSE_TIMEOUT):
     """
     Collect responses from user_data until the final response flag is seen.
     """
     errors = []
     responses = []
     while True:
         try:
-            result = user_data._response_queue.get(timeout=DEFAULT_RESPONSE_TIMEOUT)
+            result = user_data._response_queue.get(timeout=timeout)
         except queue.Empty:
-            raise Exception(
-                f"No response received within {DEFAULT_RESPONSE_TIMEOUT} seconds."
-            )
+            raise Exception(f"No response received within {timeout} seconds.")
 
         if isinstance(result, InferenceServerException):
             errors.append(result)
@@ -486,5 +487,68 @@ def test_step2_max_queue_size(self):
         self._run_inference(model_name=model_name, expected_responses_count=32)
 
 
+class EnsembleParallelFailedEnqueueTest(tu.TestResultCollector):
+    def _run_inference(self, expected_responses_count=32):
+        """
+        Exercise a fan-out ensemble where one parallel branch hits queue-full
+        first. Successful responses emitted before the failure should still be
+        correct, and the stream should terminate with exactly one queue-full
+        error.
+        """
+        user_data = UserData()
+        with grpcclient.InferenceServerClient(SERVER_URL) as triton_client:
+            try:
+                inputs, outputs = prepare_infer_args(expected_responses_count)
+                triton_client.start_stream(callback=partial(callback, user_data))
+                triton_client.async_stream_infer(
+                    model_name=MODEL_ENSEMBLE_PARALLEL_FAILED_ENQUEUE,
+                    inputs=inputs,
+                    outputs=outputs,
+                )
+
+                errors, responses = collect_responses(user_data, timeout=15)
+                self.assertLess(
+                    len(responses),
+                    expected_responses_count,
+                    "Expected the parallel slow branch to queue-fill before all "
+                    "responses completed.",
+                )
+                self.assertEqual(
+                    len(errors),
+                    1,
+                    "Expected exactly one queue-full error from the parallel "
+                    "failed-enqueue path.",
+                )
+                self.assertIn(
+                    "Exceeds maximum queue size",
+                    str(errors[0]),
+                    f"Expected queue size error, got: {str(errors[0])}",
+                )
+
+                for idx, resp in enumerate(responses):
+                    output = resp.as_numpy("OUT")
+                    self.assertAlmostEqual(
+                        float(np.squeeze(output)),
+                        EXPECTED_PARALLEL_FAILED_ENQUEUE_OUTPUT,
+                        places=5,
+                        msg=f"Response {idx} has incorrect value - {output}",
+                    )
+            finally:
+                triton_client.stop_stream()
+
+    def test_parallel_step_failed_enqueue(self):
+        """
+        Repeat the same request according to PARALLEL_FAILED_ENQUEUE_LOOPS.
+        """
+        loop_count = int(os.environ.get("PARALLEL_FAILED_ENQUEUE_LOOPS", "1"))
+        self.assertGreaterEqual(
+            loop_count, 1, "PARALLEL_FAILED_ENQUEUE_LOOPS must be >= 1"
+        )
+
+        for iteration in range(loop_count):
+            with self.subTest(iteration=iteration):
+                self._run_inference()
+
+
 if __name__ == "__main__":
     unittest.main()

qa/L0_simple_ensemble/test.sh

@@ -241,6 +241,186 @@ kill $SERVER_PID
 wait $SERVER_PID
 
 
+######## Test parallel-step failed enqueue path in ensemble scheduler ########
+PARALLEL_FAILED_ENQUEUE_MODEL_DIR="`pwd`/parallel_failed_enqueue_test_models"
+rm -rf ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}
+
+mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/1
+mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1
+mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1
+mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1
+mkdir -p ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1
+
+# Producer emits repeated responses with a larger payload value so the
+# queue-limited branch fills first.
+cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/1/model.py \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1
+cp ${BACKPRESSURE_TEST_MODEL_DIR}/decoupled_producer/config.pbtxt \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/
+sed -i 's/name: "decoupled_producer"/name: "decoupled_producer_parallel_queue"/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/config.pbtxt
+sed -i 's/0.5/2.0/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/decoupled_producer_parallel_queue/1/model.py
+
+# Queue-limited branch used to trigger a failed enqueue.
+cp ../python_models/ground_truth/model.py \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/1
+cp ../python_models/ground_truth/config.pbtxt \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/
+sed -i 's/name: "ground_truth"/name: "slow_consumer_queue_limited"/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt
+sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt
+cat >> ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/slow_consumer_queue_limited/config.pbtxt << 'EOF'
+
+dynamic_batching {
+  preferred_batch_size: [ 1 ]
+  default_queue_policy {
+    max_queue_size: 1
+  }
+}
+EOF
+
+# Parallel branch with the same interface and no added delay.
+cp ../python_models/ground_truth/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1
+cp ../python_models/ground_truth/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/
+sed -i 's/name: "ground_truth"/name: "fast_consumer"/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt
+sed -i 's/max_batch_size: 64/max_batch_size: 1/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/config.pbtxt
+sed -i 's/time.sleep(delay)/time.sleep(0)/g' \
+  ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/fast_consumer/1/model.py
+
+# Join both parallel branches into the ensemble output.
+cp ../python_models/join_add_sub/model.py ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/1
+cp ../python_models/join_add_sub/config.pbtxt ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/join_add_sub/
+
+cat > ${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}/ensemble_parallel_step_failed_enqueue/config.pbtxt << 'EOF'
+name: "ensemble_parallel_step_failed_enqueue"
+platform: "ensemble"
+max_batch_size: 0
+
+input [
+  {
+    name: "IN"
+    data_type: TYPE_INT32
+    dims: [ 1 ]
+  }
+]
+
+output [
+  {
+    name: "OUT"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+
+ensemble_scheduling {
+  step [
+    {
+      model_name: "decoupled_producer_parallel_queue"
+      model_version: -1
+      input_map {
+        key: "IN"
+        value: "IN"
+      }
+      output_map {
+        key: "OUT"
+        value: "intermediate"
+      }
+    },
+    {
+      model_name: "slow_consumer_queue_limited"
+      model_version: -1
+      input_map {
+        key: "INPUT0"
+        value: "intermediate"
+      }
+      output_map {
+        key: "OUTPUT0"
+        value: "slow_out"
+      }
+    },
+    {
+      model_name: "fast_consumer"
+      model_version: -1
+      input_map {
+        key: "INPUT0"
+        value: "intermediate"
+      }
+      output_map {
+        key: "OUTPUT0"
+        value: "fast_out"
+      }
+    },
+    {
+      model_name: "join_add_sub"
+      model_version: -1
+      input_map {
+        key: "INPUT0"
+        value: "slow_out"
+      }
+      input_map {
+        key: "INPUT1"
+        value: "fast_out"
+      }
+      output_map {
+        key: "OUTPUT0"
+        value: "OUT"
+      }
+    }
+  ]
+}
+EOF
+
+BACKPRESSURE_TEST_PY=./ensemble_backpressure_test.py
+TEST_NAME="EnsembleParallelFailedEnqueueTest.test_parallel_step_failed_enqueue"
+SERVER_LOG="./ensemble_parallel_failed_enqueue_test_server.log"
+CLIENT_LOG="./ensemble_parallel_failed_enqueue_test_client.log"
+rm -f $SERVER_LOG $CLIENT_LOG
+
+SERVER_ARGS="--model-repository=${PARALLEL_FAILED_ENQUEUE_MODEL_DIR}"
+run_server
+if [ "$SERVER_PID" == "0" ]; then
+    echo -e "\n***\n*** Failed to start $SERVER\n***"
+    cat $SERVER_LOG
+    exit 1
+fi
+
+set +e
+PARALLEL_FAILED_ENQUEUE_LOOPS=${PARALLEL_FAILED_ENQUEUE_LOOPS:-1} \
+python $BACKPRESSURE_TEST_PY $TEST_NAME -v >> $CLIENT_LOG 2>&1
+if [ $? -ne 0 ]; then
+    RET=1
+    cat $CLIENT_LOG
+else
+    check_test_results $TEST_RESULT_FILE 1
+    if [ $? -ne 0 ]; then
+        cat $CLIENT_LOG
+        echo -e "\n***\n*** Test Result Verification Failed\n***"
+        RET=1
+    fi
+fi
+
+if ! kill -0 $SERVER_PID > /dev/null 2>&1; then
+    cat $SERVER_LOG
+    echo -e "\n***\n*** Server exited during parallel failed enqueue test\n***"
+    RET=1
+else
+    wait_for_server_live $SERVER_PID 5
+    if [ "$WAIT_RET" != "0" ]; then
+        cat $SERVER_LOG
+        echo -e "\n***\n*** Server did not remain live after parallel failed enqueue test\n***"
+        RET=1
+    fi
+fi
+set -e
+
+kill $SERVER_PID > /dev/null 2>&1 || true
+wait $SERVER_PID > /dev/null 2>&1 || true
+
+
 ######## Test backpressure feature - 'max_inflight_requests' config option ########
 ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR="`pwd`/ensemble_backpressure_test_models"
 rm -rf ${ENSEMBLE_BACKPRESSURE_TEST_MODEL_DIR}

qa/python_models/join_add_sub/config.pbtxt

@@ -0,0 +1,60 @@
+# Copyright 2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions
+# are met:
+#  * Redistributions of source code must retain the above copyright
+#    notice, this list of conditions and the following disclaimer.
+#  * Redistributions in binary form must reproduce the above copyright
+#    notice, this list of conditions and the following disclaimer in the
+#    documentation and/or other materials provided with the distribution.
+#  * Neither the name of NVIDIA CORPORATION nor the names of its
+#    contributors may be used to endorse or promote products derived
+#    from this software without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS ``AS IS'' AND ANY
+# EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+# PURPOSE ARE DISCLAIMED.  IN NO EVENT SHALL THE COPYRIGHT OWNER OR
+# CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+# EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+# PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
+# PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY
+# OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+name: "join_add_sub"
+backend: "python"
+max_batch_size: 1
+
+input [
+  {
+    name: "INPUT0"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+input [
+  {
+    name: "INPUT1"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+output [
+  {
+    name: "OUTPUT0"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+output [
+  {
+    name: "OUTPUT1"
+    data_type: TYPE_FP32
+    dims: [ 1 ]
+  }
+]
+
+instance_group [{ kind: KIND_CPU }]