feat: add --max-suite-retries option to cap total reruns across suite

CHANGES.rst

@@ -4,7 +4,13 @@ Changelog
 16.5 (unreleased)
 -----------------
 
-- Nothing changed yet.
+Features
+++++++++
+
+- Add ``--max-suite-reruns`` option to cap the total number of reruns across
+  the entire test suite. Once the limit is reached, no further reruns occur
+  regardless of per-test ``--reruns`` or ``@pytest.mark.flaky`` settings.
+  Fixes `#298 <https://github.com/pytest-dev/pytest-rerunfailures/issues/298>`_.
 
 
 16.4 (2026-07-01)

README.rst

@@ -238,6 +238,22 @@ setting. To make them additive instead, pass ``--reruns-mode=append``. With
 
    $ pytest --reruns 4 --reruns-mode append
 
+Limit total reruns across the suite
+------------------------------------
+
+To cap the total number of reruns across the entire test suite regardless of
+how many individual tests fail, pass ``--max-suite-reruns``. Once the limit
+is reached, no further reruns occur even if individual tests have remaining
+reruns:
+
+.. code-block:: bash
+
+   $ pytest --reruns 3 --max-suite-reruns 10
+
+This is useful in large test suites to bound resource usage when many tests
+are flaky at the same time. The cap applies after rerun selection, including
+tests configured with ``--force-reruns`` and ``@pytest.mark.flaky``.
+
 Show tracebacks for retried failures
 ------------------------------------
 

pyproject.toml

@@ -47,6 +47,11 @@ urls = { Homepage = "https://github.com/pytest-dev/pytest-rerunfailures" }
 
 entry-points.pytest11.rerunfailures = "pytest_rerunfailures"
 
+[dependency-groups]
+dev = [
+  "mypy>=2.1",
+]
+
 [tool.setuptools.dynamic]
 readme = { file = [ "HEADER.rst", "README.rst", "CHANGES.rst" ] }
 
@@ -72,3 +77,12 @@ lint.pydocstyle.convention = "google"
 
 [tool.check-manifest]
 ignore = [ ".pre-commit-config.yaml" ]
+
+[tool.mypy]
+python_version = "3.10"
+warn_return_any = true
+warn_unused_configs = true
+
+[[tool.mypy.overrides]]
+module = [ "xdist.newhooks" ]
+ignore_missing_imports = true

src/pytest_rerunfailures.py

@@ -10,19 +10,27 @@
 import traceback
 import warnings
 from contextlib import suppress
+from typing import Any
 
 import pytest
 from _pytest.outcomes import fail
 from _pytest.runner import runtestprotocol
 from packaging.version import parse as parse_version
 
+failed_subtests_key: Any
+SubtestReport: Any
+_failed_subtests_key: Any = None
+_SubtestReport: Any = None
+
 try:
-    from _pytest.subtests import SubtestReport, failed_subtests_key
+    from _pytest.subtests import SubtestReport as _SubtestReport
+    from _pytest.subtests import failed_subtests_key as _failed_subtests_key
 except ImportError:
     if pytest.version_tuple >= (9, 0, 0):
         raise
-    failed_subtests_key = None
-    SubtestReport = None
+
+failed_subtests_key = _failed_subtests_key
+SubtestReport = _SubtestReport
 
 try:
     from xdist.newhooks import pytest_handlecrashitem
@@ -140,6 +148,15 @@ def pytest_addoption(parser):
         "'rerun test summary info' section, which is emitted automatically "
         "when this flag is set.",
     )
+    group.addoption(
+        "--max-suite-reruns",
+        action="store",
+        dest="max_suite_reruns",
+        type=int,
+        default=None,
+        help="Maximum total number of reruns across the entire test suite. "
+        "Once this limit is reached, no further reruns will occur.",
+    )
 
     arg_type = "string"
     parser.addini("reruns", RERUNS_DESC, type=arg_type)
@@ -155,10 +172,14 @@ def pytest_addoption(parser):
 # should run before / at the beginning of pytest_cmdline_main
 def check_options(config):
     val = config.getvalue
-    if not val("collectonly"):
-        if config.option.reruns != 0:
-            if config.option.usepdb:  # a core option
-                raise pytest.UsageError("--reruns incompatible with --pdb")
+    if (
+        config.option.max_suite_reruns is not None
+        and config.option.max_suite_reruns < 0
+    ):
+        raise pytest.UsageError("--max-suite-reruns must be >= 0")
+    if not val("collectonly") and config.option.reruns != 0:
+        if config.option.usepdb:  # a core option
+            raise pytest.UsageError("--reruns incompatible with --pdb")
 
 
 def _get_marker(item):
@@ -528,6 +549,7 @@ def pytest_configure(config):
         "seconds between re-runs, multiplied by 'reruns_delay_backoff_factor' "
         "after each attempt for an exponential backoff.",
     )
+    check_options(config)
 
     if config.pluginmanager.hasplugin("xdist") and HAS_PYTEST_HANDLECRASHITEM:
         config.pluginmanager.register(XDistHooks())
@@ -548,11 +570,23 @@ def pytest_handlecrashitem(self, crashitem, report, sched):
         """Return the crashitem from pending and collection."""
         db = sched.config.failures_db
         reruns = db.get_test_reruns(crashitem)
+        reserved_suite_rerun = False
         if db.get_test_failures(crashitem) < reruns:
+            max_suite_reruns = sched.config.option.max_suite_reruns
+            if max_suite_reruns is None:
+                cap_available = True
+            else:
+                reserved_suite_rerun = db.try_increment_suite_reruns(max_suite_reruns)
+                cap_available = reserved_suite_rerun
+        else:
+            cap_available = False
+        if cap_available:
             try:
                 sched.mark_test_pending(crashitem)
                 report.outcome = "rerun"
             except NotImplementedError:
+                if reserved_suite_rerun:
+                    db.decrement_suite_reruns()
                 # Some schedulers (like LoadScopeScheduling) don't implement
                 # mark_test_pending
                 # In this case, we can't reschedule the crashed test for rerun
@@ -576,9 +610,39 @@ def pytest_handlecrashitem(self, crashitem, report, sched):
 # and failures (set after each failure or crash)
 # accessible from both the master and worker
 class StatusDB:
-    def __init__(self):
-        self.delim = b"\n"
-        self.hmap = {}
+    def __init__(self) -> None:
+        self.delim: bytes = b"\n"
+        self.hmap: dict[str, str] = {}
+        self._suite_rerun_count: int = 0
+        self._suite_lock = threading.Lock()
+
+    def increment_suite_reruns(self) -> int:
+        """Atomically increment the suite-wide rerun counter; return new total."""
+        with self._suite_lock:
+            self._suite_rerun_count += 1
+            return self._suite_rerun_count
+
+    def try_increment_suite_reruns(self, max_cap: int) -> bool:
+        """Increment the suite counter when it is below the configured cap."""
+        with self._suite_lock:
+            if self._suite_rerun_count < max_cap:
+                self._suite_rerun_count += 1
+                return True
+            return False
+
+    def decrement_suite_reruns(self) -> None:
+        """Release a suite slot when a scheduled rerun cannot be started."""
+        with self._suite_lock:
+            if self._suite_rerun_count > 0:
+                self._suite_rerun_count -= 1
+
+    def get_suite_reruns(self) -> int:
+        """Return the current suite-wide rerun count.
+
+        Reads under lock for thread safety.
+        """
+        with self._suite_lock:
+            return self._suite_rerun_count
 
     def _hash(self, crashitem: str) -> str:
         if crashitem not in self.hmap:
@@ -635,12 +699,12 @@ def _sock_send(self, conn, msg: str):
 
 
 class ServerStatusDB(SocketDB):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__()
         self.sock.bind(("127.0.0.1", 0))
         self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
 
-        self.rerunfailures_db = {}
+        self.rerunfailures_db: dict[str, dict[str, int]] = {}
         t = threading.Thread(target=self.run_server, daemon=True)
         t.start()
 
@@ -663,6 +727,25 @@ def run_connection(self, conn):
                     self._set(i, k, int(v))
                 elif op == "get":
                     self._sock_send(conn, str(self._get(i, k)))
+                elif op == "inc":
+                    with self._suite_lock:
+                        new_v = self._get(i, k) + 1
+                        self._set(i, k, new_v)
+                    self._sock_send(conn, str(new_v))
+                elif op == "try_inc":
+                    with self._suite_lock:
+                        current = self._get(i, k)
+                        if current < int(v):
+                            self._set(i, k, current + 1)
+                            self._sock_send(conn, "1")
+                        else:
+                            self._sock_send(conn, "0")
+                elif op == "dec":
+                    with self._suite_lock:
+                        current = self._get(i, k)
+                        if current > 0:
+                            self._set(i, k, current - 1)
+                    self._sock_send(conn, "1")
 
     def _set(self, i: str, k: str, v: int):
         if i not in self.rerunfailures_db:
@@ -675,6 +758,33 @@ def _get(self, i: str, k: str) -> int:
         except KeyError:
             return 0
 
+    def increment_suite_reruns(self) -> int:
+        """Atomically increment the suite-wide rerun counter; return new total."""
+        with self._suite_lock:
+            new_v = self._get("__suite__", "r") + 1
+            self._set("__suite__", "r", new_v)
+            return new_v
+
+    def try_increment_suite_reruns(self, max_cap: int) -> bool:
+        """Increment the suite counter when it is below the configured cap."""
+        with self._suite_lock:
+            current = self._get("__suite__", "r")
+            if current < max_cap:
+                self._set("__suite__", "r", current + 1)
+                return True
+            return False
+
+    def decrement_suite_reruns(self) -> None:
+        """Release a suite slot when a scheduled rerun cannot be started."""
+        with self._suite_lock:
+            current = self._get("__suite__", "r")
+            if current > 0:
+                self._set("__suite__", "r", current - 1)
+
+    def get_suite_reruns(self) -> int:
+        """Return the current suite-wide rerun count."""
+        return self._get("__suite__", "r")
+
 
 class ClientStatusDB(SocketDB):
     def __init__(self, sock_port):
@@ -688,8 +798,35 @@ def _get(self, i: str, k: str) -> int:
         self._sock_send(self.sock, "|".join(("get", i, k, "")))
         return int(self._sock_recv(self.sock))
 
+    def increment_suite_reruns(self) -> int:
+        """Atomically increment the suite-wide rerun counter; return new total."""
+        self._sock_send(self.sock, "|".join(("inc", "__suite__", "r", "")))
+        return int(self._sock_recv(self.sock))
+
+    def try_increment_suite_reruns(self, max_cap: int) -> bool:
+        """Increment the suite counter when it is below the configured cap."""
+        self._sock_send(
+            self.sock, "|".join(("try_inc", "__suite__", "r", str(max_cap)))
+        )
+        return self._sock_recv(self.sock) == "1"
+
+    def decrement_suite_reruns(self) -> None:
+        """Release a suite slot when a scheduled rerun cannot be started."""
+        self._sock_send(self.sock, "|".join(("dec", "__suite__", "r", "")))
+        self._sock_recv(self.sock)
+
+    def get_suite_reruns(self) -> int:
+        """Return the current suite-wide rerun count."""
+        return self._get("__suite__", "r")
 
-suspended_finalizers = {}
+
+suspended_finalizers: dict[Any, Any] = {}
+
+
+def _restore_suspended_finalizers(item):
+    """Restore higher-scope finalizers after a rerun is not scheduled."""
+    item.session._setupstate.stack.update(suspended_finalizers)
+    suspended_finalizers.clear()
 
 
 def pytest_runtest_teardown(item, nextitem):
@@ -706,6 +843,14 @@ def pytest_runtest_teardown(item, nextitem):
 
     _test_failed_statuses = getattr(item, "_test_failed_statuses", {})
 
+    max_suite_reruns = item.session.config.option.max_suite_reruns
+    if (
+        max_suite_reruns is not None
+        and item.session.config.failures_db.get_suite_reruns() >= max_suite_reruns
+    ):
+        _restore_suspended_finalizers(item)
+        return
+
     # Only remove non-function level actions from the stack if the test is to be re-run
     # Exceeding re-run limits, being free of failue statuses, and encountering
     # allowable exceptions indicate that the test is not to be re-ran.
@@ -726,8 +871,7 @@ def pytest_runtest_teardown(item, nextitem):
                     del item.session._setupstate.stack[key]
     else:
         # restore suspended finalizers
-        item.session._setupstate.stack.update(suspended_finalizers)
-        suspended_finalizers.clear()
+        _restore_suspended_finalizers(item)
 
 
 @pytest.hookimpl(hookwrapper=True)
@@ -762,9 +906,6 @@ def pytest_runtest_protocol(item, nextitem):
         # flaky
         return
 
-    # while this doesn't need to be run with every item, it will fail on the
-    # first item if necessary
-    check_options(item.session.config)
     delay = get_reruns_delay(item)
     delay_backoff_factor = get_reruns_delay_backoff_factor(item)
     parallel = not is_master(item.config)
@@ -788,6 +929,14 @@ def pytest_runtest_protocol(item, nextitem):
                 item.ihook.pytest_runtest_logreport(report=report)
             else:
                 # failure detected and reruns not exhausted, since i < reruns
+                max_suite_reruns = item.session.config.option.max_suite_reruns
+                if max_suite_reruns is not None:
+                    if not db.try_increment_suite_reruns(max_suite_reruns):
+                        # Suite-wide limit exhausted -- log as final failure.
+                        _restore_suspended_finalizers(item)
+                        item.ihook.pytest_runtest_logreport(report=report)
+                        continue
+
                 report.outcome = "rerun"
                 time.sleep(delay * delay_backoff_factor ** (item.execution_count - 1))