From 687bee282e62f3cf675ae7def35e9582716b963e Mon Sep 17 00:00:00 2001
From: fcostaoliveira <filipe@redis.com>
Date: Mon, 26 Oct 2026 00:28:49 +0000
Subject: [PATCH] Don't let a slow queue feeder make the runner skip tests

The serial runner put every test on a multiprocessing.Queue and read it
back with get(timeout=0.1), stopping on Empty. Queue.put() is flushed by
a background feeder thread, so when that thread was starved for 100 ms
the runner ran zero tests and exited 0.

Iterate the test list directly in the serial path. In the parallel path,
end the queue with one None sentinel per worker and use a blocking get().

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 RLTest/__main__.py                     | 28 +++++------
 tests/unit/test_serial_runner_queue.py | 70 ++++++++++++++++++++++++++
 2 files changed, 83 insertions(+), 15 deletions(-)
 create mode 100644 tests/unit/test_serial_runner_queue.py

diff --git a/RLTest/__main__.py b/RLTest/__main__.py
index e6a5a4a..4db0a07 100644
--- a/RLTest/__main__.py
+++ b/RLTest/__main__.py
@@ -994,20 +994,14 @@ def execute(self):
             print(Colors.Bred('only one test can be run on interactive-debugger use -t'))
             sys.exit(1)
 
-        jobs = Queue()
-        n_jobs = 0
-        for test in self.loader:
-            jobs.put(test, block=False)
-            n_jobs += 1
+        tests = list(self.loader)
+        n_jobs = len(tests)
 
-        def run_jobs_main_thread(jobs):
+        def run_jobs_main_thread(tests):
             nonlocal done
             bar = self.progressbar(n_jobs)
-            for _ in bar:
-                try:
-                    test = jobs.get(timeout=0.1)
-                except Exception as e:
-                    break
+            for i in bar:
+                test = tests[i]
 
                 def on_timeout():
                     nonlocal done
@@ -1027,9 +1021,8 @@ def on_timeout():
         def run_jobs(jobs, results, port):
             Defaults.port = port
             while True:
-                try:
-                    test = jobs.get(timeout=0.1)
-                except Exception as e:
+                test = jobs.get()
+                if test is None:
                     break
 
                 # Reset per-test: addFailure() in this worker writes into this
@@ -1081,8 +1074,13 @@ def on_timeout():
         # Open group for all tests at the start (parallel execution)
         self._openGitHubActionsTestsGroup()
         if self.parallelism == 1:
-            run_jobs_main_thread(jobs)
+            run_jobs_main_thread(tests)
         else :
+            # A timed get() can see an empty queue before the feeder thread
+            # flushes, so workers block on get() until their None sentinel.
+            jobs = Queue()
+            for test in tests + [None] * self.parallelism:
+                jobs.put(test, block=False)
             processes = []
             currPort = Defaults.port
             for i in range(self.parallelism):
diff --git a/tests/unit/test_serial_runner_queue.py b/tests/unit/test_serial_runner_queue.py
new file mode 100644
index 0000000..29330b1
--- /dev/null
+++ b/tests/unit/test_serial_runner_queue.py
@@ -0,0 +1,70 @@
+"""Regression test: the serial runner must run every test even when a
+multiprocessing.Queue's feeder thread has not flushed yet.
+
+It used to read tests back from a multiprocessing.Queue with
+get(timeout=0.1) and stop on Empty, so a starved feeder thread made it run
+zero tests and exit 0.
+"""
+
+import argparse
+import multiprocessing as mp
+import queue
+import sys
+from types import SimpleNamespace
+from unittest import TestCase
+from unittest.mock import patch
+
+import RLTest.__main__ as rlmain
+from RLTest.__main__ import RLTest
+
+
+class _StarvedQueue:
+    """A queue whose feeder thread never gets to flush."""
+
+    def put(self, obj, block=True, timeout=None):
+        pass
+
+    def get(self, block=True, timeout=None):
+        raise queue.Empty
+
+
+def _make_rltest(tests, parallelism, run_single_test):
+    rl = RLTest.__new__(RLTest)
+    rl.args = argparse.Namespace(env_only=False, interactive_debugger=False,
+                                 no_output_catch=True, no_progress=True,
+                                 failed_tests_file=None)
+    rl.loader = tests
+    rl.parallelism = parallelism
+    rl.currEnv = None
+    rl.testsFailed = {}
+    rl.github_actions_group_open = False
+    rl.run_single_test = run_single_test
+    return rl
+
+
+class TestRunnerQueue(TestCase):
+
+    def test_serial_runs_every_test_with_a_starved_queue(self):
+        ran = []
+
+        def run_single_test(test, on_timeout):
+            ran.append(test)
+            return 1
+
+        rl = _make_rltest(['t1', 't2', 't3'], 1, run_single_test)
+        with patch.object(rlmain, 'Queue', _StarvedQueue):
+            rl.execute()
+        self.assertEqual(ran, ['t1', 't2', 't3'])
+
+    def test_parallel_runs_every_test(self):
+        if sys.platform == 'win32':
+            self.skipTest('fork start method is unavailable on Windows')
+        ctx = mp.get_context('fork')
+        tests = [SimpleNamespace(name='t%d' % i) for i in range(20)]
+        rl = _make_rltest(tests, 3, lambda test, on_timeout: 1)
+        with patch.object(rlmain, 'Queue', ctx.Queue), \
+             patch.object(rlmain, 'Process', ctx.Process), \
+             patch('sys.stdout') as out:
+            rl.execute()
+        summary = ''.join(c.args[0] for c in out.write.call_args_list)
+        self.assertIn('Total Tests Run: 20,', summary)