This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-8049-e87ec126afdbdb12a9fc90e4904e54dc9a59c7ca in repository https://gitbox.apache.org/repos/asf/texera.git
commit 1d47f1d8642a29fcb857ba10fe425aa12e043c90 Author: Xinyuan Lin <[email protected]> AuthorDate: Fri Aug 28 02:53:57 2026 +0000 test(pyamber): cover the statistics and pause managers (#8049) ### What changes were proposed in this PR? Two manager specs extended, 19 tests to 27. | File | Codecov fully-covered | Branch arms | |---|---|---| | `statistics_manager.py` | 41/45 → **45/45** | 12/14 → **14/14** | | `pause_manager.py` | 42/45 → **45/45** | 12/14 → **14/14** | **+7 fully-covered lines and +4 branch arms. Both files reach 100% on lines and branches.** `statistics_manager.py`'s gap was `94: 50% (1/2)`, `95: MISS`, `101: 50% (1/2)`, `102: MISS` — the arms deciding whether an input or output statistics update is recorded at all. `pause_manager.py`'s was the channel pause/resume paths, including both `change_state=False` variants and the global-pause queue cycle. **Two files were dropped from this bundle on measurement:** `state_manager.py` (1 line) and `linked_blocking_multi_queue.py`, whose four partial lines did not survive scrutiny despite its 78 branch arms. `core/util/customized_queue/inner.py` sits in the same package and shows 2 partials plus 5 missed lines, but it is **vendored third-party code** and was refused rather than counted. ### Verification Two reviewers returned **eleven** findings — the most of any bundle in this batch. The builder had reported `survivors: []`, which the reviewers showed was materially wrong. **46 mutations were run one at a time against the final tree; 44 die, and 15 of those die at exactly one test each.** The two that survive are reported rather than dropped, both genuinely equivalent: - Exchanging the `.add(channel_id)` and `disable(channel_id)` statements inside `pause_input_channel` — both statements are individually pinned by separate tests, so the order is unobservable. - Replacing the guarded `remove` with `discard` — `discard` is by definition `remove if present`. Measured from `amber/` with the identical full-suite scope on both sides and `--cov-branch` passed explicitly, since this branch is on `main` and `amber/pyproject.toml` has no `[tool.coverage.run]` section (verified by reading it: only the three `ruff` tables and `pytest.ini_options` exist). Per-line `hits` and `condition-coverage` were read out of `coverage.xml` by a parser script; summary percentages ignored. The before-state was produced by writing the `HEAD` content of the two test files into the tree from a scratch extraction and restoring the new versions from a scratch snapshot afterwards, sha256-verified. **No `git restore` or `git checkout` was run on any file.** Full pyamber suite: `5 failed, 1167 passed, 7 errors` against main's `1159 passed`. Failure sets sorted and diffed — **identical by identity**, 12 entries. `ruff check` and `ruff format --check` pass on CI's exact scope. No production file is touched; `git diff -- 'amber/src/main/*'` is empty. ### Any related issues, documentation, discussions? Closes #8045 ### How was this PR tested? ``` cd amber && python -m pytest -m "not integration" -q src/test/python/core/architecture/managers ``` ``` 27 passed ``` ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5) --------- Signed-off-by: Xinyuan Lin <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../architecture/managers/test_pause_manager.py | 135 +++++++++++++++++++++ .../managers/test_statistics_manager.py | 95 ++++++++++++++- 2 files changed, 229 insertions(+), 1 deletion(-) diff --git a/amber/src/test/python/core/architecture/managers/test_pause_manager.py b/amber/src/test/python/core/architecture/managers/test_pause_manager.py index 501c133500..d006e3f247 100644 --- a/amber/src/test/python/core/architecture/managers/test_pause_manager.py +++ b/amber/src/test/python/core/architecture/managers/test_pause_manager.py @@ -20,6 +20,9 @@ import pytest 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 @@ class TestPauseManager: 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() diff --git a/amber/src/test/python/core/architecture/managers/test_statistics_manager.py b/amber/src/test/python/core/architecture/managers/test_statistics_manager.py index 5abf7a36b4..3c88f850ba 100644 --- a/amber/src/test/python/core/architecture/managers/test_statistics_manager.py +++ b/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 @@ class TestStatisticsManagerExecutionTime: 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
