test(pyamber): cover the statistics and pause managers

amber/src/test/python/core/architecture/managers/test_pause_manager.py

@@ -20,6 +20,9 @@
 from core.architecture.managers import StateManager
 from core.architecture.managers.pause_manager import PauseManager, PauseType
 from core.models import InternalQueue
+from core.models.internal_queue import DataElement
+from core.models.payload import DataPayload
+from proto.org.apache.texera.amber.core import ActorVirtualIdentity, ChannelIdentity
 from proto.org.apache.texera.amber.engine.architecture.worker import WorkerState
 
 
@@ -72,3 +75,135 @@ def test_it_can_be_resumed_when_resumed(self, pause_manager):
         assert not pause_manager.is_paused()
         pause_manager.resume(PauseType.USER_PAUSE)
         assert not pause_manager.is_paused()
+
+    @staticmethod
+    def _register_channel(input_queue, name: str) -> ChannelIdentity:
+        """A channel only gets a sub-queue once something is enqueued on it, so
+        go through the production put() path before pausing it."""
+        channel = ChannelIdentity(
+            ActorVirtualIdentity(name), ActorVirtualIdentity("self"), False
+        )
+        input_queue.put(DataElement(tag=channel, payload=DataPayload()))
+        return channel
+
+    def test_resume_of_one_type_keeps_worker_paused_while_another_is_held(
+        self, pause_manager, state_manager
+    ):
+        # Two independent global pause holders. Releasing one must not resume
+        # the worker -- resume() returns early while any global pause remains.
+        pause_manager.pause(PauseType.USER_PAUSE)
+        pause_manager.pause(PauseType.DEBUG_PAUSE)
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+
+        pause_manager.resume(PauseType.USER_PAUSE)
+        assert pause_manager.is_paused()
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+
+        # resume() must release the type it was ASKED for, not just some
+        # arbitrary holder: repeating the release of the already-released
+        # USER_PAUSE is a no-op, and DEBUG_PAUSE still holds the worker.
+        # Without this, `resume(X)` popping an arbitrary element off the pause
+        # set is indistinguishable from popping X -- and an EXCEPTION_PAUSE
+        # released by a user resume would silently restart a failed worker.
+        pause_manager.resume(PauseType.USER_PAUSE)
+        assert pause_manager.is_paused()
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+
+        # Releasing the last holder does resume.
+        pause_manager.resume(PauseType.DEBUG_PAUSE)
+        assert not pause_manager.is_paused()
+        assert state_manager.confirm_state(WorkerState.RUNNING)
+
+    def test_global_pause_closes_the_data_queue_and_resume_reopens_it(
+        self, pause_manager, input_queue
+    ):
+        # pause()/resume() gate the data queues wholesale, keyed by
+        # DISABLE_BY_PAUSE. The channel is registered BEFORE the pause, which
+        # is what makes the blanket disable observable at all: disable_data()
+        # only iterates the sub-queues that exist when it runs, so a test that
+        # registers afterwards cannot see it. Re-enabling under the wrong
+        # DisableType key would leave a resumed worker's data queue shut for
+        # good, i.e. a hang.
+        self._register_channel(input_queue, "up")
+        assert input_queue.is_data_enabled()
+
+        pause_manager.pause(PauseType.USER_PAUSE)
+        assert not input_queue.is_data_enabled()
+
+        pause_manager.resume(PauseType.USER_PAUSE)
+        assert input_queue.is_data_enabled()
+
+    def test_channel_pause_blocks_the_global_resume(
+        self, pause_manager, state_manager, input_queue
+    ):
+        # A per-channel pause outlives the global one: once the global pause is
+        # released the pause set is empty, but a channel is still held, so the
+        # blanket data re-enable and the PAUSED -> RUNNING transition are both
+        # skipped. Note the channel is registered AFTER pause(), so pause()'s
+        # own disable_data() found no sub-queues and did nothing here -- the
+        # channel is closed only by pause_input_channel below.
+        pause_manager.pause(PauseType.USER_PAUSE)
+        channel = self._register_channel(input_queue, "up")
+        pause_manager.pause_input_channel(PauseType.DEBUG_PAUSE, channel)
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+        assert not input_queue.is_data_enabled()
+
+        pause_manager.resume(PauseType.USER_PAUSE)
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+        # The blanket re-enable really is skipped, not merely state-neutral.
+        assert not input_queue.is_data_enabled()
+
+        # Releasing the channel pause finally resumes.
+        pause_manager.resume(PauseType.DEBUG_PAUSE)
+        assert state_manager.confirm_state(WorkerState.RUNNING)
+        assert input_queue.is_data_enabled()
+
+    def test_releasing_one_channel_pause_reopens_that_channel(
+        self, pause_manager, input_queue
+    ):
+        # Two channels held by two DIFFERENT pause types, and no global pause.
+        # Releasing one type leaves the other still holding, so resume()
+        # returns before the blanket enable_data() at the end of the method --
+        # which means the released channel can only be open again because
+        # resume() re-enabled it channel by channel. In the sibling test above,
+        # that blanket enable masks the per-channel work entirely.
+        first = self._register_channel(input_queue, "first")
+        second = self._register_channel(input_queue, "second")
+        pause_manager.pause_input_channel(PauseType.DEBUG_PAUSE, first)
+        # ECM_PAUSE is the type main_loop actually uses for channel pauses.
+        pause_manager.pause_input_channel(PauseType.ECM_PAUSE, second)
+        assert not input_queue.is_data_enabled()
+        assert not input_queue._queue.is_enabled(first)
+        assert not input_queue._queue.is_enabled(second)
+
+        pause_manager.resume(PauseType.DEBUG_PAUSE)
+        assert input_queue._queue.is_enabled(first)
+        assert not input_queue._queue.is_enabled(second)
+
+    def test_pause_with_change_state_false_leaves_the_state_alone(
+        self, pause_manager, state_manager
+    ):
+        # change_state=False is the "gate the queues but do not touch the
+        # reported worker state" path. No production caller overrides the
+        # default today, so this pins the parameter's contract rather than a
+        # live behaviour -- but without it the guard is vacuously true and
+        # deleting `change_state and` from either method changes nothing.
+        pause_manager.pause(PauseType.USER_PAUSE, change_state=False)
+        assert state_manager.confirm_state(WorkerState.READY)
+
+    def test_resume_with_change_state_false_reopens_the_queue_but_not_the_state(
+        self, pause_manager, state_manager, input_queue
+    ):
+        self._register_channel(input_queue, "up")
+        pause_manager.pause(PauseType.USER_PAUSE)
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+        assert not input_queue.is_data_enabled()
+
+        pause_manager.resume(PauseType.USER_PAUSE, change_state=False)
+        # The state label deliberately lags...
+        assert state_manager.confirm_state(WorkerState.PAUSED)
+        # ...but the worker really is running again: no pause holder remains
+        # and the data queue is open. is_paused() must follow the pause set,
+        # not the stale label, or a caller polling it would wait forever.
+        assert input_queue.is_data_enabled()
+        assert not pause_manager.is_paused()

amber/src/test/python/core/architecture/managers/test_statistics_manager.py

@@ -16,6 +16,7 @@
 # under the License.
 
 import pytest
+from loguru import logger
 
 from core.architecture.managers.statistics_manager import StatisticsManager
 from proto.org.apache.texera.amber.core import PortIdentity
@@ -138,10 +139,102 @@ def test_total_execution_time_before_start_raises(self):
     def test_idle_time_clamped_to_zero_when_processing_overshoots(self):
         # When data+control exceed total_execution_time (e.g. update_total was
         # called before all increase_* calls for that interval), idle_time is
-        # clamped to 0 and a warning is logged. It must never be negative.
+        # clamped to 0. It must never be negative.
+        # No drift warning fires HERE, despite the shape of the scenario: the
+        # increase_* calls come after update_total, so processing_total was
+        # still 0 when the guard ran. The warning path needs the opposite
+        # order and is covered in TestStatisticsManagerDriftWarnings.
         mgr = StatisticsManager()
         mgr.initialize_worker_start_time(1_000)
         mgr.update_total_execution_time(1_100)  # 100ns total
         mgr.increase_data_processing_time(80)
         mgr.increase_control_processing_time(50)  # 130 > 100
         assert mgr.get_statistics().idle_time == 0
+
+
+def _capture(call) -> list:
+    """Run `call` with a record-capturing loguru sink attached and return the
+    records. The sink is attached at DEBUG, not WARNING, so that a diagnostic
+    silently DEMOTED below WARNING stays distinguishable from one that is gone
+    -- a WARNING-level sink cannot tell those two apart. loguru's logger is a
+    process-global singleton and the pytest process is shared, so the removal
+    has to happen in a finally or a leaked sink poisons every sibling suite.
+    (loguru does not propagate to stdlib logging, so caplog is not an option.)
+    """
+    records: list = []
+    handler_id = logger.add(lambda m: records.append(m.record), level="DEBUG")
+    try:
+        call()
+    finally:
+        logger.remove(handler_id)
+    return records
+
+
+def _messages(call) -> list:
+    return [r["message"] for r in _capture(call)]
+
+
+class TestStatisticsManagerDriftWarnings:
+    """The two warning paths in update_total_execution_time. Both are pure
+    diagnostics -- the value is still stored -- so the assertions pin the
+    message CONTENT and the log LEVEL, otherwise swapping the two warning
+    bodies, or escalating one to ERROR, would survive."""
+
+    def test_non_monotonic_total_execution_time_warns_and_still_stores(self):
+        mgr = StatisticsManager()
+        # The worker start time and the stored total are deliberately DIFFERENT
+        # literals (100 vs 1000). Were they equal, the message assertion below
+        # could be satisfied by _worker_start_time standing in for
+        # _total_execution_time, and would then pin neither field.
+        mgr.initialize_worker_start_time(100)
+        mgr.update_total_execution_time(1_100)  # total_execution_time = 1000
+
+        # new_total = 500 < stored 1000 -> clock went backwards.
+        records = _capture(lambda: mgr.update_total_execution_time(600))
+
+        joined = "".join(r["message"] for r in records)
+        assert "non-monotonic time" in joined
+        assert "new total 500ns < current total 1000ns" in joined
+        # Not the other warning: 500 >= data(0) + control(0).
+        assert "idle_time drift" not in joined
+        # A defensive diagnostic against clock skew, not an alert: exactly one
+        # record, and it stays at WARNING.
+        assert [r["level"].name for r in records] == ["WARNING"]
+        # Last write still wins -- the warning does not veto the update.
+        assert mgr.get_statistics().idle_time == 500
+
+        # Boundary: the guard is `<`, so re-sending the SAME timestamp is
+        # monotonic and must stay silent. main_loop calls this repeatedly, so
+        # `<=` here would warn on every unchanged timestamp.
+        assert _messages(lambda: mgr.update_total_execution_time(600)) == []
+        assert mgr.get_statistics().idle_time == 500
+
+    def test_idle_drift_warns_naming_data_and_control_totals(self):
+        mgr = StatisticsManager()
+        mgr.initialize_worker_start_time(1_000)
+        mgr.increase_data_processing_time(80)
+        mgr.increase_control_processing_time(50)  # processing_total = 130
+
+        # new_total = 100 < 130 -> idle_time would go negative.
+        records = _capture(lambda: mgr.update_total_execution_time(1_100))
+
+        joined = "".join(r["message"] for r in records)
+        assert "idle_time drift" in joined
+        assert "total_execution_time (100ns) < data (80ns) + control (50ns)" in joined
+        # Not the other warning: 100 >= stored total 0.
+        assert "non-monotonic time" not in joined
+        assert [r["level"].name for r in records] == ["WARNING"]
+        assert mgr.get_statistics().idle_time == 0
+
+        # No false alarm on the boundary. A fresh manager whose total lands
+        # EXACTLY on data+control has idle_time 0, not negative, so neither
+        # warning may fire. This pins both operands of the comparison as well
+        # as the boundary: comparing the stored total instead of new_total, or
+        # multiplying data by control instead of adding them, each makes this
+        # block warn.
+        mgr2 = StatisticsManager()
+        mgr2.initialize_worker_start_time(1_000)
+        mgr2.increase_data_processing_time(80)
+        mgr2.increase_control_processing_time(50)
+        assert _messages(lambda: mgr2.update_total_execution_time(1_130)) == []
+        assert mgr2.get_statistics().idle_time == 0