From 28e1353e0c203460aee8084f70f3a479d35ade2e Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 21 Aug 2026 14:51:37 +0200 Subject: [PATCH] fix: wait for the Leios voting committee epoch The Leios voting committee is drawn from a stake distribution snapshot that stays empty until epoch 3 on a freshly started cluster instance. Until then every pool answers each EB announcement with `NotOnCommittee`, so no EB can be voted on or certified. `test_eb_logs` searched for votes and certificates from the moment it started, which on a fresh instance was always before epoch 3, so the vote and certificate messages could never show up. Wait for the epoch in which the committee becomes active, and schedule the test at the end of the testrun so the wait is normally a no-op. Add `test_no_voting_before_committee_epoch`, scheduled near the start of the testrun, to check the other side of the same behavior: no vote and no certificate is reported before epoch 3, while the pools do report that they are not on the committee. The searched log window must not cross an epoch boundary, otherwise a vote from epoch 3 could land in it, so `wait_for_epoch_interval` positions the search early enough in an epoch for the whole window to fit into it. The epoch and the position in it are read from a single tip before that wait, so the test skips right away instead of burning a whole epoch when no epoch before the voting one has room left. The epoch is read again after the wait and after the search, as only the second read tells which epoch the search ran in. --- cardano_node_tests/tests/test_leios_blocks.py | 323 +++++++++++++++--- 1 file changed, 280 insertions(+), 43 deletions(-) diff --git a/cardano_node_tests/tests/test_leios_blocks.py b/cardano_node_tests/tests/test_leios_blocks.py index 6e2b12ccb..210739554 100644 --- a/cardano_node_tests/tests/test_leios_blocks.py +++ b/cardano_node_tests/tests/test_leios_blocks.py @@ -19,6 +19,7 @@ from cardano_node_tests.tests import common from cardano_node_tests.utils import cluster_nodes +from cardano_node_tests.utils import clusterlib_utils from cardano_node_tests.utils import configuration from cardano_node_tests.utils import helpers from cardano_node_tests.utils import logfiles @@ -40,6 +41,16 @@ pytest.mark.skipif(bool(_SKIP_REASON), reason=_SKIP_REASON), ] +# EB messages that a node can report only once the Leios voting committee is active. +# They belong to more than one of the groups below, so they are named here. +_MSG_VOTE_ACQUIRED = r"Consensus\.LeiosKernel\.VoteAcquired\].*Leios vote acquired" +_MSG_VOTED = r"Consensus\.LeiosKernel\.Voted\].*Leios voted, weight=" +_MSG_CERTIFIED = r"Consensus\.LeiosKernel\.Certified\].*Leios cert assembled for RB" +_MSG_BLOCK_CERTIFIED = r"Consensus\.LeiosKernel\.BlockCertified\].*EB certified at slot" +_MSG_CERTIFIED_AND_ANNOUNCED = ( + r"Consensus\.LeiosKernel\.CertifiedAndAnnounced\].*RB certified an EB and announced a new one" +) + # EB messages that every block producing node takes part in. Every pool sees every EB # announcement and every vote, and votes on and downloads the EBs forged by the other # pools. The peer side messages need an EB forged by another pool, so they would be @@ -51,9 +62,9 @@ r"Consensus\.LeiosKernel\.AnnouncementAccepted\].*EB announcement accepted from", r"Consensus\.LeiosKernel\.BlockAcquired\].*EB body acquired:", r"Consensus\.LeiosKernel\.BlockTxsAcquired\].*EB txs acquired:", - r"Consensus\.LeiosKernel\.VoteAcquired\].*Leios vote acquired", - r"Consensus\.LeiosKernel\.Voted\].*Leios voted, weight=", - r"Consensus\.LeiosKernel\.Certified\].*Leios cert assembled for RB", + _MSG_VOTE_ACQUIRED, + _MSG_VOTED, + _MSG_CERTIFIED, ) # EB messages reported only by the pool that won the EB election, so they are expected @@ -62,13 +73,31 @@ r"Consensus\.LeiosKernel\.BlockForged\].*EB forged at slot", r"Consensus\.LeiosKernel\.BlockAnnounced\].*EB announced:", r"Consensus\.LeiosKernel\.BlockStored\].*EB stored at slot", - r"Consensus\.LeiosKernel\.BlockCertified\].*EB certified at slot", - r"Consensus\.LeiosKernel\.CertifiedAndAnnounced\].*RB certified an EB and announced a new one", + _MSG_BLOCK_CERTIFIED, + _MSG_CERTIFIED_AND_ANNOUNCED, ) -_EB_MSGS_COMPILED = {r: re.compile(r) for r in EB_MSGS_ALL_POOLS + EB_MSGS_ANY_POOL} +# EB messages that prove the voting committee is active - a vote was cast, received or +# turned into a certificate. +VOTING_MSGS = ( + _MSG_VOTE_ACQUIRED, + _MSG_VOTED, + _MSG_CERTIFIED, + _MSG_BLOCK_CERTIFIED, + _MSG_CERTIFIED_AND_ANNOUNCED, +) + +# Reported by a pool for an EB announcement it cannot vote on, because it is not a +# member of the voting committee for the EB. +NOT_ON_COMMITTEE_MSG = r"Consensus\.LeiosKernel\.NotVoted\].*Leios not voted for .*: NotOnCommittee" + +_EB_MSGS_COMPILED = { + r: re.compile(r) for r in (*EB_MSGS_ALL_POOLS, *EB_MSGS_ANY_POOL, NOT_ON_COMMITTEE_MSG) +} _ALL_POOLS_MSGS_SET = frozenset(EB_MSGS_ALL_POOLS) _ANY_POOL_MSGS_SET = frozenset(EB_MSGS_ANY_POOL) +_VOTING_MSGS_SET = frozenset(VOTING_MSGS) +_PRE_VOTING_MSGS_SET = frozenset((*VOTING_MSGS, NOT_ON_COMMITTEE_MSG)) # Max number of new blocks to wait for while the expected messages are showing up in # the logs. Certification of an EB needs a quorum of votes and an RB that announces it, @@ -86,6 +115,28 @@ # rate of the `local_fast` variant (and its 4x shorter epoch) drains the mempool. MIN_BLOCK_INTERVAL_SEC = 10 +# The first epoch in which a pool can be a member of the Leios voting committee. The +# committee is drawn from a stake distribution snapshot that is empty for the whole +# lifetime of a freshly started cluster instance until this epoch, so up to then every +# pool answers every EB announcement with `NotOnCommittee` and no EB can be voted on or +# certified. +VOTING_START_EPOCH = 3 + +# Number of seconds between two searches of the logs while waiting for the pools to +# report that they are not on the voting committee. +NO_VOTE_SEARCH_STEP_SEC = 30 + +# Max number of seconds to spend searching the logs for the pools reporting that they +# are not on the voting committee. EBs come in bursts, with gaps of up to ~250 sec +# between them on `leios_fast`, so 360 sec is one full gap plus slack - long enough for +# at least one burst to normally land in the window. +MAX_NO_VOTE_SEARCH_SEC = 360 + +# Number of seconds to keep between the end of the searched log window and the end of +# the epoch it runs in, so that a vote logged in the next epoch cannot land in the +# searched part of the log. +VOTING_EPOCH_MARGIN_SEC = 60 + @pytest.fixture def cluster_leios(cluster: clusterlib.ClusterLib) -> clusterlib.ClusterLib: @@ -194,6 +245,64 @@ def _get_missing_msgs_errors(*, found_per_pool: dict[pl.Path, set[str]]) -> list return errors +def _init_searches(pool_logs: list[pl.Path]) -> dict[pl.Path, _LogSearch]: + """Return the search state for each pool log, starting at its current end.""" + searches: dict[pl.Path, _LogSearch] = {} + for logfile in pool_logs: + seek_offset, inode, timestamp = _get_log_position(logfile) + searches[logfile] = _LogSearch(seek_offset=seek_offset, inode=inode, timestamp=timestamp) + return searches + + +def _search_round( + *, + searches: dict[pl.Path, _LogSearch], + missing_msgs: tp.Callable[[_LogSearch], tp.Collection[str]], + log_errors: dict[pl.Path, str], +) -> None: + """Search the part of each pool log that was appended since the previous round. + + Updates `searches` in place with what was found and where to continue, and + `log_errors` with the log files that could not be searched. + + Args: + searches: The search state per pool log. + missing_msgs: Returns the messages a given search state is still missing, so the + caller decides what the round looks for. + log_errors: Only the last outcome per log file, so that repeated rotations don't + pile up near-duplicate reports - the error names the rotated file the search + died on. An entry is removed once a later round recovered from the failure. + """ + for logfile, search in searches.items(): + missing = missing_msgs(search) + if not missing: + continue + + try: + # Record the new search position before the search, so that lines appended + # while the search is running are not skipped in the next round + next_position = _get_log_position(logfile) + search.found |= _find_eb_msgs( + regexes=missing, + logfile=logfile, + seek_offset=search.seek_offset, + inode=search.inode, + timestamp=search.timestamp, + ) + except FileNotFoundError as err: + # The log file kept getting rotated during the search. Keep the search + # position, so that the same part of the log is searched again. + msg = f"Cannot search '{logfile}': {err}" + LOGGER.warning("%s", msg) + log_errors[logfile] = msg + continue + + # The search position was kept on failure, so a failure that a later round + # recovered from didn't cost any log content + log_errors.pop(logfile, None) + search.seek_offset, search.inode, search.timestamp = next_position + + def _collect_eb_msgs( *, cluster_obj: clusterlib.ClusterLib, pool_logs: list[pl.Path] ) -> tuple[dict[pl.Path, set[str]], list[str]]: @@ -212,16 +321,14 @@ def _collect_eb_msgs( the search - a log file that could not be searched, or a stalled chain. They explain a missing message, so they are reported only together with one. """ - searches: dict[pl.Path, _LogSearch] = {} - for logfile in pool_logs: - seek_offset, inode, timestamp = _get_log_position(logfile) - searches[logfile] = _LogSearch(seek_offset=seek_offset, inode=inode, timestamp=timestamp) - + searches = _init_searches(pool_logs) stall_error = "" - # Only the last outcome per log file, so that repeated rotations don't pile up - # near-duplicate reports - the error names the rotated file the search died on log_errors: dict[pl.Path, str] = {} + def _missing_msgs(search: _LogSearch) -> tp.Collection[str]: + found_any_pool = {m for s in searches.values() for m in s.found} + return (_ALL_POOLS_MSGS_SET - search.found) | (_ANY_POOL_MSGS_SET - found_any_pool) + for __ in range(MAX_WAIT_BLOCKS // WAIT_BLOCKS_STEP): try: cluster_obj.wait_for_new_block(new_blocks=WAIT_BLOCKS_STEP) @@ -230,36 +337,7 @@ def _collect_eb_msgs( # report what was still missing and not just the stall itself. stall_error = f"The chain stalled while waiting for new blocks: {err}" - found_any_pool = {m for s in searches.values() for m in s.found} - for logfile, search in searches.items(): - missing = (_ALL_POOLS_MSGS_SET - search.found) | (_ANY_POOL_MSGS_SET - found_any_pool) - if not missing: - continue - - try: - # Record the new search position before the search, so that lines - # appended while the search is running are not skipped in the next round - next_position = _get_log_position(logfile) - search.found |= _find_eb_msgs( - regexes=missing, - logfile=logfile, - seek_offset=search.seek_offset, - inode=search.inode, - timestamp=search.timestamp, - ) - except FileNotFoundError as err: - # The log file kept getting rotated during the search. Keep the search - # position, so that the same part of the log is searched again. - msg = f"Cannot search '{logfile}': {err}" - LOGGER.warning("%s", msg) - log_errors[logfile] = msg - continue - - # The search position was kept on failure, so a failure that a later round - # recovered from didn't cost any log content - log_errors.pop(logfile, None) - search.seek_offset, search.inode, search.timestamp = next_position - found_any_pool |= search.found + _search_round(searches=searches, missing_msgs=_missing_msgs, log_errors=log_errors) found_per_pool = {p: s.found for p, s in searches.items()} if stall_error or not _get_missing_msgs_errors(found_per_pool=found_per_pool): @@ -272,10 +350,57 @@ def _collect_eb_msgs( return {p: s.found for p, s in searches.items()}, problems +def _collect_pre_voting_msgs( + *, pool_logs: list[pl.Path], deadline: float +) -> tuple[dict[pl.Path, set[str]], list[str]]: + """Search the pool logs for voting activity while the voting committee is empty. + + The search stops as soon as every pool log holds a `NotOnCommittee` message, as that + is the proof that the pool could not vote, or as soon as any voting message shows up, + as that is what the caller reports. Each round searches only the part of a log that + was appended since the previous round. + + Args: + pool_logs: Log files of the block producing nodes. + deadline: A `time.monotonic()` value the search must not go past, so that the + searched part of the logs stays inside the epoch the search started in. + + Returns: + The searched messages found in each pool log, and the problems that got in the + way of the search - a log file that could not be searched. They explain a missing + message, so they are reported only together with one. + """ + searches = _init_searches(pool_logs) + log_errors: dict[pl.Path, str] = {} + + while True: + time.sleep(min(NO_VOTE_SEARCH_STEP_SEC, max(0.0, deadline - time.monotonic()))) + + _search_round( + searches=searches, + missing_msgs=lambda search: _PRE_VOTING_MSGS_SET - search.found, + log_errors=log_errors, + ) + + # Voting activity is what the caller reports, no need to keep searching for it + if any(s.found & _VOTING_MSGS_SET for s in searches.values()): + break + # Every pool reported that it is not a member of the voting committee + if all(NOT_ON_COMMITTEE_MSG in s.found for s in searches.values()): + break + if time.monotonic() >= deadline: + break + + return {p: s.found for p, s in searches.items()}, list(log_errors.values()) + + class TestLeios: """Tests for Leios endorser blocks.""" @allure.link(helpers.get_vcs_link()) + # Scheduled at the end of the testrun, so that the cluster instance is already past + # `VOTING_START_EPOCH` and the wait for it is a no-op + @pytest.mark.order(-10) @pytest.mark.long def test_eb_logs( self, @@ -283,6 +408,7 @@ def test_eb_logs( ): """Check that the nodes report the expected endorser block activity. + * Wait for the epoch in which the voting committee becomes active * Record the current end of each pool log file * Wait for new blocks to be created * Check that each pool reports EB announcements, votes and certificates @@ -292,6 +418,10 @@ def test_eb_logs( cluster = cluster_leios common.get_test_id(cluster) + # Votes and certificates cannot show up in the logs before the voting committee + # is active, so searching for them earlier would always fail + cluster.wait_for_epoch(epoch_no=VOTING_START_EPOCH, padding_seconds=5) + state_dir = cluster_nodes.get_cluster_env().state_dir pool_logs = sorted(state_dir.glob("pool*.stdout")) assert pool_logs, f"No pool log files found in '{state_dir}'" @@ -303,3 +433,110 @@ def test_eb_logs( errors.extend(search_problems) assert not errors, "\n".join(errors) + + @allure.link(helpers.get_vcs_link()) + # Scheduled near the start of the testrun, while the cluster instance is still + # before `VOTING_START_EPOCH` + @pytest.mark.order(5) + @pytest.mark.long + def test_no_voting_before_committee_epoch( + self, + cluster_leios: clusterlib.ClusterLib, + ): + """Check that no EB is voted on before the voting committee becomes active. + + * Skip when no epoch before `VOTING_START_EPOCH` has room left for the whole + search window + * Wait for a point in an epoch where the window fits before the next epoch + boundary + * Record the current end of each pool log file + * Search the log content that gets appended, until every pool reports that it + declined to vote because it is not a member of the voting committee + * Check that no pool reported a vote, a vote from a peer or a certificate + """ + cluster = cluster_leios + common.get_test_id(cluster) + + state_dir = cluster_nodes.get_cluster_env().state_dir + pool_logs = sorted(state_dir.glob("pool*.stdout")) + assert pool_logs, f"No pool log files found in '{state_dir}'" + + # The searched log window must not cross an epoch boundary, otherwise a vote from + # `VOTING_START_EPOCH` could land in it. It fits into an epoch only when it starts + # at least `epoch_tail_sec` before the end of that epoch. + epoch_tail_sec = MAX_NO_VOTE_SEARCH_SEC + VOTING_EPOCH_MARGIN_SEC + if cluster.epoch_length_sec <= epoch_tail_sec: + pytest.skip( + f"An epoch takes only {cluster.epoch_length_sec:.0f} sec on the " + f"'{configuration.TESTNET_VARIANT}' testnet variant, which is not enough for " + f"the {epoch_tail_sec} sec search window" + ) + + # One tip for both values, so that the epoch cannot flip between them. Check + # before waiting for the interval, as that wait can take a whole epoch. + tip = cluster.g_query.get_tip() + init_epoch = int(tip["epoch"]) + last_usable_epoch = VOTING_START_EPOCH - 1 + if init_epoch > last_usable_epoch or ( + init_epoch == last_usable_epoch + and cluster.time_from_epoch_start(tip=tip) > cluster.epoch_length_sec - epoch_tail_sec + ): + pytest.skip( + f"The cluster instance is in epoch {init_epoch} and no epoch before " + f"{VOTING_START_EPOCH}, in which the Leios voting committee becomes active, " + f"has {epoch_tail_sec} sec left for the search window" + ) + + clusterlib_utils.wait_for_epoch_interval(cluster_obj=cluster, start=0, stop=-epoch_tail_sec) + + # The wait can cross into the next epoch, so the epoch the search runs in is not + # necessarily the one seen above + search_epoch = cluster.g_query.get_epoch() + if search_epoch >= VOTING_START_EPOCH: + pytest.skip( + f"The cluster instance is already in epoch {search_epoch}, the Leios voting " + f"committee is active since epoch {VOTING_START_EPOCH}" + ) + + found_per_pool, search_problems = _collect_pre_voting_msgs( + pool_logs=pool_logs, deadline=time.monotonic() + MAX_NO_VOTE_SEARCH_SEC + ) + + # A vote is legitimate from `VOTING_START_EPOCH` on, so the result means nothing + # if the searched window reached that epoch after all + end_epoch = cluster.g_query.get_epoch() + if end_epoch >= VOTING_START_EPOCH: + pytest.skip( + f"The search started in epoch {search_epoch} and ended in epoch {end_epoch}, " + "in which the Leios voting committee is active, so the result is inconclusive" + ) + + errors = [ + f"Found a line matching `{r}` in '{logfile}' in epoch {search_epoch}." + for logfile, found in found_per_pool.items() + for r in VOTING_MSGS + if r in found + ] + if errors: + errors.extend(search_problems) + + assert not errors, "\n".join(errors) + + # Without a pool declining to vote there was nothing to vote on, so the absence + # of votes doesn't say anything about the voting committee + no_evidence = sorted( + str(logfile) + for logfile, found in found_per_pool.items() + if NOT_ON_COMMITTEE_MSG not in found + ) + if no_evidence: + pytest.skip( + "; ".join( + [ + "No pool declined to vote with `NotOnCommittee` in " + f"{', '.join(no_evidence)}, so the absence of votes in epoch " + f"{search_epoch} is inconclusive", + *search_problems, + ] + ) + )