From f7cd29372d10a98ec6829a7caa5537bf37f9a63e Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Mon, 9 Nov 2026 16:08:34 +0100 Subject: [PATCH] fix(tests): harden dynamic block production Skip unsafe forecast windows before funding or reconfiguration, accounting for the producing stake lost when a pool becomes a backup. Preserve both BLS voting identities on the backup and use explicit epoch targets with boundary buffers and validated ledger snapshots. Pause firehose during restarts and redirect it to the surviving relay. Validated with lint, 551 passing framework tests, the leios_fast skip, and a passing local_fast E2E run with 15 TPS load and clean teardown. --- cardano_node_tests/tests/test_blocks.py | 228 ++++++++++++++++-------- 1 file changed, 151 insertions(+), 77 deletions(-) diff --git a/cardano_node_tests/tests/test_blocks.py b/cardano_node_tests/tests/test_blocks.py index 1fee4913b..6186903f0 100644 --- a/cardano_node_tests/tests/test_blocks.py +++ b/cardano_node_tests/tests/test_blocks.py @@ -3,8 +3,10 @@ Other block production checks may be present in `test_staking.py`. """ +import json import logging import os +import pathlib as pl import random import re import shutil @@ -419,10 +421,53 @@ class TestDynamicBlockProd: producing_node_name = "pool1" backup_node_name = "pool2" + @pytest.fixture + def cluster_dynamic( + self, + cluster_manager: cluster_management.ClusterManager, + cluster_singleton: clusterlib.ClusterLib, + ) -> clusterlib.ClusterLib: + """Lock the cluster and skip unsafe forecast windows before funding addresses. + + Taking the backup pool's forging identity offline reduces the live stake. Require + the probability of no leader in one forecast window to be at most 1e-9, using the + Praos empty-slot probability `(1 - f) ** live_stake`. Otherwise one unlucky gap + can exhaust the forecast and permanently stop this isolated cluster's forging. + """ + cluster = cluster_singleton + # Up to 1.5 epochs to reach the reconfiguration interval, its remainder, + # and three full observation epochs. + common.skip_on_long_epochs(cluster_obj=cluster, epochs=5) + backup_pool_id = delegation.get_pool_id( + cluster_obj=cluster, + addrs_data=cluster_manager.cache.addrs_data, + pool_name=f"node-{self.backup_node_name}", + ) + stake_distribution = cluster.g_query.get_stake_distribution() + # Bootstrap delegations may not be active yet. The local cluster gives each + # pool an equal stake share; do not underestimate the stake lost later on. + backup_stake = max(stake_distribution[backup_pool_id], 1 / configuration.NUM_POOLS) + live_stake = 1 - backup_stake + forecast_slots = clusterlib_utils.get_stability_window(cluster_obj=cluster) + empty_window_probability = (1 - float(cluster.genesis["activeSlotsCoeff"])) ** ( + live_stake * forecast_slots + ) + if empty_window_probability > 1e-9: + pytest.skip( + f"Unsafe forecast window for dynamic block production: {forecast_slots} slots, " + f"{live_stake:.1%} producing stake, probability of no leader " + f"{empty_window_probability:.3g} exceeds 1e-9" + ) + return cluster + def reconf_for_dynamic(self) -> None: """Reconfigure cluster for dynamic block production. Reconfigure nodeX to be backup block production node for nodeY. + On Leios, retain both pools' BLS voting keys on the backup so taking either + forging process offline does not remove a committee seat. + Pause firehose across the restart and move its submission socket off the node + that will be stopped during failover. Stop all nodes before applying the Supervisor configuration, then start them once. """ state_dir = cluster_nodes.get_cluster_env().state_dir @@ -434,11 +479,27 @@ def reconf_for_dynamic(self) -> None: LOGGER.info( f"Reconfiguring node '{self.backup_node_name}' to be non-producing backup node" - f"for '{self.producing_node_name}'." + f" for '{self.producing_node_name}'." ) shutil.move(backup_node_dir, backup_node_dir_orig) shutil.copytree(producing_node_dir, backup_node_dir) + # Voting runs independently of forging and is not reloaded by SIGHUP. Load both + # identities at restart, preserving any existing rotation-key bundles as well. + bls_skey = backup_node_dir / "bls.skey" + original_bls_skey = backup_node_dir_orig / "bls.skey" + if bls_skey.exists() and original_bls_skey.exists(): + voting_keys: list[dict[str, str]] = [] + for key_file in (bls_skey, original_bls_skey): + with open(key_file, encoding="utf-8") as in_fp: + keys = json.load(in_fp) + for key in keys if isinstance(keys, list) else [keys]: + if key not in voting_keys: + voting_keys.append(key) + with open(bls_skey, "w", encoding="utf-8") as out_fp: + json.dump(voting_keys, out_fp, indent=4) + out_fp.write("\n") + with open(supervisor_conf, encoding="utf-8") as fp_in: supervisor_conf_content = fp_in.read() @@ -450,6 +511,18 @@ def reconf_for_dynamic(self) -> None: with open(supervisor_conf, "w", encoding="utf-8") as fp_out: fp_out.write(supervisor_conf_content) + # Firehose cannot stay connected across the restart or the permanent stop of + # the producing node. Keep the load running through failover via the relay. + if configuration.ENABLE_TX_FIREHOSE: + cluster_nodes.services_action(service_names=["tx_firehose"], action="stop") + firehose_config_file = state_dir / "tx-firehose-config.json" + with open(firehose_config_file, encoding="utf-8") as in_fp: + firehose_config = json.load(in_fp) + socket_path = pl.Path(firehose_config["socket_path"]) + if socket_path.name == f"{self.producing_node_name}.socket": + firehose_config["socket_path"] = str(socket_path.with_name("bft1.socket")) + helpers.write_json(out_file=firehose_config_file, content=firehose_config) + # All nodes must restart together on local testnets. Updating a running group # and then restarting it would interrupt the nodes during startup. cluster_nodes.stop_nodes(["*"]) @@ -457,15 +530,17 @@ def reconf_for_dynamic(self) -> None: # Nodes auto-started by update are left running, not restarted by start. cluster_nodes.start_nodes(["*"]) time.sleep(configuration.TX_SUBMISSION_DELAY) + if configuration.ENABLE_TX_FIREHOSE: + cluster_nodes.services_action(service_names=["tx_firehose"], action="start") @pytest.fixture def payment_addrs( self, cluster_manager: cluster_management.ClusterManager, - cluster_singleton: clusterlib.ClusterLib, + cluster_dynamic: clusterlib.ClusterLib, ) -> list[clusterlib.AddressRecord]: """Create new payment addresses.""" - cluster = cluster_singleton + cluster = cluster_dynamic addrs = addrs_common.get_payment_addrs( name_template=common.get_test_id(cluster), cluster_manager=cluster_manager, @@ -480,7 +555,7 @@ def payment_addrs( def test_dynamic_block_production( self, cluster_manager: cluster_management.ClusterManager, - cluster_singleton: clusterlib.ClusterLib, + cluster_dynamic: clusterlib.ClusterLib, payment_addrs: list[clusterlib.AddressRecord], worker_id: str, ): @@ -488,13 +563,12 @@ def test_dynamic_block_production( * Check that blocks are produced by both nodeX and nodeY pools * Reconfigure a nodeX to be non-producing backup node for nodeY pool + * Preserve both pools' Leios voting identities, when present * Terminate nodeY and send SIGHUP to nodeX so it starts producing blocks on behalf of nodeY + * Target each epoch explicitly and fail if an observation epoch is missed * Check that nodeX has replaced nodeY and is producing blocks on its behalf """ - cluster = cluster_singleton - # Waits for up to 5 epochs: up to 1.5 to reach the second half of an epoch in - # epoch 1+, the rest of that epoch, and `num_epochs` more epochs - common.skip_on_long_epochs(cluster_obj=cluster, epochs=5) + cluster = cluster_dynamic temp_template = common.get_test_id(cluster) num_epochs = 3 @@ -520,6 +594,10 @@ def _save_state(curr_epoch: int) -> dict[str, int]: state_name=f"{temp_template}_epoch{curr_epoch}", ledger_state=ledger_state, ) + assert ledger_state["lastEpoch"] == curr_epoch, ( + f"Missed observation epoch {curr_epoch}: ledger is in epoch " + f"{ledger_state['lastEpoch']}" + ) blocks_before: dict[str, int] = ledger_state["blocksBefore"] return blocks_before @@ -550,78 +628,74 @@ def _save_state(curr_epoch: int) -> dict[str, int]: "Failed to finish reconfiguration in single epoch, it would affect other checks" ) - epoch_end = cluster.time_to_epoch_end(tip) - curr_time = time.monotonic() - epoch_end_deadline = curr_time + epoch_end - test_end_deadline = epoch_end_deadline + (num_epochs * cluster.epoch_length_sec) - - blocks_db = {} + # Leave a transaction-sized margin before boundaries. A slow submission may + # still cross one, so wait for absolute targets rather than "the next epoch". + stop_buffer = common.get_epoch_stop_sec_buffer(cluster_obj=cluster) + blocks_before: dict[str, int] = {} LOGGER.info(f"Checking blocks for {num_epochs} epochs.") - while curr_time < test_end_deadline: - epoch_end = epoch_end_deadline - curr_time - if epoch_end < 15: - LOGGER.info(f"End of epoch {curr_epoch}, saving data.") - blocks_db[curr_epoch] = _save_state(curr_epoch) - - if curr_epoch == reconf_epoch: - assert backup_pool_id in blocks_db[curr_epoch], ( - "The original pool should have forged blocks before it became backup node." - ) - assert producing_pool_id in blocks_db[curr_epoch], ( - "Producing pool should forge blocks." - ) - - curr_epoch = cluster.wait_for_new_epoch(padding_seconds=5) - epoch_end_deadline = cluster.time_to_epoch_end() + time.monotonic() - - # Replace the node - if curr_epoch == reconf_epoch + 1: - LOGGER.info( - f"Replacing node '{self.producing_node_name}' " - f"with '{self.backup_node_name}'." - ) - backup_node_status = cluster_nodes.services_status( - service_names=[f"nodes:{self.backup_node_name}"] - )[0] - if not backup_node_status.pid: - msg = "Replacement node is not running." - raise AssertionError(msg) - # Sending `SIGHUP` triggers the node's RPC configuration reload, which - # re-parses the node config file and fails with - # "Error while updating RPC configuration: Missing YAML config file". - # The reload re-validates the whole config from the file alone, but the - # config file path itself is only supplied via the `--config` CLI argument - # and is never stored in the file, so the validation always fails. The - # failure is non-fatal (the reload is simply skipped), so ignore the error. - # See https://github.com/IntersectMBO/cardano-node/issues/6589 - logfiles.add_ignore_rule( - files_glob=f"{self.backup_node_name}.stdout", - regex="RpcConfigUpdateError", - ignore_file_id=worker_id, - ) - cluster_nodes.stop_nodes(node_names=[self.producing_node_name]) - os.kill(backup_node_status.pid, signal.SIGHUP) - - # Send tx - src_addr, dst_addr = random.sample(payment_addrs, 2) - txouts = [clusterlib.TxOut(address=dst_addr.address, amount=1_000_000)] - tx_files = clusterlib.TxFiles(signing_key_files=[src_addr.skey_file]) - - cluster.g_transaction.send_tx( - src_address=src_addr.address, - tx_name=f"{temp_template}_{int(time.time())}", - txouts=txouts, - tx_files=tx_files, + for observed_epoch in range(reconf_epoch, reconf_epoch + num_epochs + 1): + tip = cluster.g_query.get_tip() + assert cluster.g_query.get_epoch(tip=tip) == observed_epoch, ( + f"Missed observation epoch {observed_epoch}: tip is {tip}" ) + epoch_end_deadline = time.monotonic() + cluster.time_to_epoch_end(tip) + tx_no = 0 + while time.monotonic() < epoch_end_deadline + stop_buffer: + src_addr, dst_addr = random.sample(payment_addrs, 2) + txouts = [clusterlib.TxOut(address=dst_addr.address, amount=1_000_000)] + tx_files = clusterlib.TxFiles(signing_key_files=[src_addr.skey_file]) + + cluster.g_transaction.send_tx( + src_address=src_addr.address, + tx_name=f"{temp_template}_epoch{observed_epoch}_tx{tx_no}", + txouts=txouts, + tx_files=tx_files, + ) + tx_no += 1 + if cluster.g_query.get_epoch() != observed_epoch: + break + time.sleep(2) + + # blocksBefore describes the completed epoch, not the current one. Save + # after the boundary so delayed queries cannot silently relabel a snapshot. + curr_epoch = cluster.wait_for_epoch( + epoch_no=observed_epoch + 1, padding_seconds=5, future_is_ok=False + ) + LOGGER.info(f"Epoch {observed_epoch} finished, saving data in epoch {curr_epoch}.") + blocks_before = _save_state(curr_epoch) - time.sleep(2) - curr_time = time.monotonic() - - # Save also data for the last epoch - curr_epoch = cluster.g_query.get_epoch() - blocks_db[curr_epoch] = _save_state(curr_epoch) + # Replace the node exactly once, after observing the reconfiguration epoch. + if observed_epoch == reconf_epoch: + assert backup_pool_id in blocks_before, ( + "The original pool should have forged blocks before it became backup node." + ) + assert producing_pool_id in blocks_before, "Producing pool should forge blocks." + LOGGER.info( + f"Replacing node '{self.producing_node_name}' with '{self.backup_node_name}'." + ) + backup_node_status = cluster_nodes.services_status( + service_names=[f"nodes:{self.backup_node_name}"] + )[0] + if not backup_node_status.pid: + msg = "Replacement node is not running." + raise AssertionError(msg) + # Sending `SIGHUP` triggers the node's RPC configuration reload, which + # re-parses the node config file and fails with + # "Error while updating RPC configuration: Missing YAML config file". + # The reload re-validates the whole config from the file alone, but the + # config file path itself is only supplied via the `--config` CLI argument + # and is never stored in the file, so the validation always fails. The + # failure is non-fatal (the reload is simply skipped), so ignore the error. + # See https://github.com/IntersectMBO/cardano-node/issues/6589 + logfiles.add_ignore_rule( + files_glob=f"{self.backup_node_name}.stdout", + regex="RpcConfigUpdateError", + ignore_file_id=worker_id, + ) + cluster_nodes.stop_nodes(node_names=[self.producing_node_name]) + os.kill(backup_node_status.pid, signal.SIGHUP) - assert backup_pool_id not in blocks_db[curr_epoch], ( + assert backup_pool_id not in blocks_before, ( "The original pool should NOT forge blocks after it became backup node." ) - assert producing_pool_id in blocks_db[curr_epoch], "Producing pool should forge blocks." + assert producing_pool_id in blocks_before, "Producing pool should forge blocks."