rptest/util: add check_consistently() safety helper

The inverse of wait_until: assert a condition holds for the whole
duration window instead of waiting for it to become true, failing fast
with AssertionError the moment it breaks.

Author: nvartolomei

tests/rptest/tests/util_test.py

@@ -0,0 +1,73 @@
+# Copyright 2026 Redpanda Data, Inc.
+#
+# Use of this software is governed by the Business Source License
+# included in the file licenses/BSL.md
+#
+# As of the Change Date specified in that file, in accordance with
+# the Business Source License, use of this software will be governed
+# by the Apache License, Version 2.0
+
+import time
+
+from ducktape.mark.resource import cluster
+from ducktape.tests.test import Test
+
+from rptest.util import check_consistently, expect_exception
+
+
+class CheckConsistentlyTest(Test):
+    @cluster(num_nodes=0)
+    def test_holds_across_window(self) -> None:
+        # A condition that stays true returns without raising, polled repeatedly.
+        calls = 0
+
+        def cond() -> bool:
+            nonlocal calls
+            calls += 1
+            return True
+
+        check_consistently(cond, duration_sec=0.1, interval_sec=0.005)
+        assert calls >= 2
+
+    @cluster(num_nodes=0)
+    def test_slow_poll_forces_a_check_past_the_deadline(self) -> None:
+        calls = 0
+
+        def cond() -> bool:
+            nonlocal calls
+            calls += 1
+            if calls == 2:
+                # Begins inside the window but returns well after it.
+                time.sleep(0.5)
+            return True
+
+        check_consistently(cond, duration_sec=0.1, interval_sec=0)
+        # The slow second poll started before the deadline, so success requires
+        # a third poll -- one that begins after the deadline.
+        assert calls == 3
+
+    @cluster(num_nodes=0)
+    def test_raises_when_condition_becomes_false(self) -> None:
+        # False from the start: fails on the first poll, before sleeping.
+        calls = 0
+
+        def false_from_start() -> bool:
+            nonlocal calls
+            calls += 1
+            return False
+
+        with expect_exception(AssertionError, lambda _: True):
+            check_consistently(false_from_start, duration_sec=5, interval_sec=0.01)
+        assert calls == 1
+
+        # Flips to false mid-window: raises as soon as it does.
+        calls = 0
+
+        def flips() -> bool:
+            nonlocal calls
+            calls += 1
+            return calls < 3
+
+        with expect_exception(AssertionError, lambda _: True):
+            check_consistently(flips, duration_sec=5, interval_sec=0.001)
+        assert calls == 3

tests/rptest/util.py

@@ -218,6 +218,58 @@ def wait_until_with_progress_check(
     ) from last_exception
 
 
+def check_consistently(
+    condition: Callable[[], bool],
+    duration_sec: float,
+    interval_sec: float = 0.1,
+    err_msg: str | Callable[[], str] = "",
+) -> None:
+    """
+    Safety check: passes only if `condition` stays true for the whole
+    `duration_sec` window, and fails fast the moment it becomes false.
+
+    The safety counterpart to `wait_until`'s liveness check: `wait_until`
+    asserts that something good eventually happens (it blocks until a
+    condition becomes true), whereas `check_consistently` asserts that nothing
+    bad happens (it blocks while the condition *remains* true, raising
+    AssertionError the moment it returns false).
+
+    `condition` is polled every `interval_sec`; the `duration_sec` window is
+    measured from after the first poll, so at least `duration_sec` elapses
+    between the first check and the last. Exceptions propagate to the caller
+    unchanged.
+
+    Note: a passing safety assertion establishes only that the guarded event
+    did not occur within `duration_sec`; it does not establish that the window
+    was long enough for that event to occur at all. Too short a `duration_sec`
+    therefore yields a meaningless pass.
+
+    To make the assertion solid, pair it with a positive control -- a companion
+    test in which, under conditions that should trigger the behavior, the event
+    is observed within the same `duration_sec`. Ideally the two are
+    parameterization of a single test rather than independently written cases,
+    so that they share setup and window and cannot silently drift out of sync.
+
+    For example, a test asserting that data is not evicted while disk headroom
+    is sufficient should be accompanied by one demonstrating that the same
+    window suffices to evict data under disk pressure.
+    """
+
+    def poll() -> None:
+        if not condition():
+            msg = err_msg() if callable(err_msg) else err_msg
+            raise AssertionError(msg or f"condition did not hold for {duration_sec}s")
+
+    poll()
+    stop = time.monotonic() + duration_sec
+    while True:
+        time.sleep(interval_sec)
+        started = time.monotonic()
+        poll()
+        if started >= stop:
+            return
+
+
 def segments_count(redpanda, topic, partition_idx):
     storage = redpanda.storage(scan_cache=False)
     topic_partitions = storage.partitions("kafka", topic)
@@ -414,7 +466,9 @@ def is_truncated():
         is_truncated,
         timeout_sec=timeout_sec,
         backoff_sec=1,
-        err_msg=lambda: f"truncation couldn't be verified for {topic=} and {target_bytes=}. last run partition_sizes={sizes}",
+        err_msg=lambda: (
+            f"truncation couldn't be verified for {topic=} and {target_bytes=}. last run partition_sizes={sizes}"
+        ),
     )
 
 
@@ -560,8 +614,10 @@ def check_throttle_rate(node):
                     return True
                 metrics = list(redpanda.metrics(node))
                 family = filter(
-                    lambda fam: fam.name
-                    == "vectorized_raft_recovery_partition_movement_assigned_bandwidth",
+                    lambda fam: (
+                        fam.name
+                        == "vectorized_raft_recovery_partition_movement_assigned_bandwidth"
+                    ),
                     metrics,
                 )
                 shard_rates = next(family).samples