From 6d597101ccc5714523dcf739c355b5060ec43bb3 Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 6 Nov 2026 18:47:05 +0100 Subject: [PATCH 1/4] perf: batch testnet cleanup transactions Batch stake re-registration, withdrawals, deregistration and DRep retirement to reduce sequential confirmation waits. Preserve reward discovery after re-registration and split rejected batches to isolate stale credentials. Batch payment UTxO queries, fund only required workers, and verify return transactions together instead of waiting a fixed block count. Add phase timings, action counts and unit coverage. Validation: 578 tests passed, 1 expected failure; lint passed. Preview runtime improvement has not yet been measured. --- cardano_node_tests/utils/testnet_cleanup.py | 529 +++++++++++++------- framework_tests/test_testnet_cleanup.py | 465 +++++++++++++++++ 2 files changed, 816 insertions(+), 178 deletions(-) create mode 100644 framework_tests/test_testnet_cleanup.py diff --git a/cardano_node_tests/utils/testnet_cleanup.py b/cardano_node_tests/utils/testnet_cleanup.py index 17253a60d..ab2e631f1 100644 --- a/cardano_node_tests/utils/testnet_cleanup.py +++ b/cardano_node_tests/utils/testnet_cleanup.py @@ -7,8 +7,10 @@ """ import concurrent.futures +import dataclasses import functools import itertools +import json import logging import pathlib as pl import random @@ -18,6 +20,7 @@ from cardano_clusterlib import clusterlib from cardano_node_tests.utils import cluster_nodes +from cardano_node_tests.utils import clusterlib_utils from cardano_node_tests.utils import defragment_utxos from cardano_node_tests.utils import helpers from cardano_node_tests.utils import tx_view @@ -26,19 +29,92 @@ TxInputGroup = list[tuple[list[clusterlib.UTXOData], pl.Path]] +# Leave ample room for transaction overhead and payment inputs. Oversized or stale +# certificate batches are split further on submission failure. +CERT_BATCH_SIZE = 20 +QUERY_BATCH_SIZE = 100 -def reregister_stake_addr( + +@dataclasses.dataclass +class _CertificateAction: + """Certificates, witnesses and balance changes for one cleanup credential.""" + + address: str + certificates: list[pl.Path] + signing_keys: list[pl.Path] + deposit: int = 0 + withdrawals: list[clusterlib.TxOut] = dataclasses.field(default_factory=list) + + +def _submit_certificate_batches( *, cluster_obj: clusterlib.ClusterLib, payment_addr: clusterlib.AddressRecord, + actions: list[_CertificateAction], + name_template: str, +) -> list[_CertificateAction]: + """Submit bounded certificate batches, isolating rejected actions by splitting. + + Keep confirmation enabled: subsequent batches spend the funding address's change. + Do not split after an uncertain confirmation failure, since the original transaction + might still be applied. Return only actions whose transactions were confirmed. + """ + successful = [] + + def _submit(batch: list[_CertificateAction], name: str) -> None: + tx_files = clusterlib.TxFiles( + certificate_files=list(itertools.chain.from_iterable(a.certificates for a in batch)), + signing_key_files=list( + dict.fromkeys( + [ + payment_addr.skey_file, + *itertools.chain.from_iterable(a.signing_keys for a in batch), + ] + ) + ), + ) + try: + cluster_obj.g_transaction.send_tx( + src_address=payment_addr.address, + tx_name=name, + tx_files=tx_files, + withdrawals=list(itertools.chain.from_iterable(a.withdrawals for a in batch)), + deposit=sum(a.deposit for a in batch), + witness_count_add=2, + ) + except clusterlib.CLIError as exc: + uncertain = "didn't make it to the chain" in str(exc) or "Failed to resubmit" in str( + exc + ) + if len(batch) == 1 or uncertain: + LOGGER.exception(f"Failed cleanup batch '{name}': {[a.address for a in batch]}") + return + LOGGER.warning(f"Splitting rejected cleanup batch '{name}' ({len(batch)} credentials)") + midpoint = len(batch) // 2 + _submit(batch[:midpoint], f"{name}_0") + _submit(batch[midpoint:], f"{name}_1") + else: + successful.extend(batch) + LOGGER.info(f"Confirmed cleanup batch '{name}' ({len(batch)} credentials)") + + for i, batch in enumerate(itertools.batched(actions, CERT_BATCH_SIZE, strict=False)): + _submit(list(batch), f"rf_{name_template}_batch{i}") + LOGGER.info(f"{name_template}: confirmed {len(successful)}/{len(actions)} credential actions") + return successful + + +def _stake_registration_action( + *, + cluster_obj: clusterlib.ClusterLib, stake_addr: clusterlib.AddressRecord, stake_addr_info: clusterlib.StakeAddrInfo, name_template: str, deposit_amt: int, -) -> clusterlib.StakeAddrInfo: - """Re-register stake address. +) -> _CertificateAction | None: + """Prepare re-registration and vote delegation when needed. - The address needs to be registered and delegated to a DRep to be able to withdraw rewards. + Preserve re-registration of unqueryable accounts so any remaining rewards can be + discovered after registration. Rewards require vote delegation before withdrawal. """ rereg_certs = [] rereg_deposit = 0 @@ -65,46 +141,25 @@ def reregister_stake_addr( ) rereg_certs.append(deleg_cert) - if rereg_certs: - tx_files_register = clusterlib.TxFiles( - certificate_files=rereg_certs, - signing_key_files=[payment_addr.skey_file, stake_addr.skey_file], - ) - try: - cluster_obj.g_transaction.send_tx( - src_address=payment_addr.address, - tx_name=f"rf_{name_template}_rereg_stake", - tx_files=tx_files_register, - deposit=rereg_deposit, - ) - except clusterlib.CLIError: - LOGGER.exception( - f"Failed to submit stake address re-registration '{stake_addr.address}'" - ) - else: - for ch in range(3): - if ch > 0: - cluster_obj.wait_for_new_block() - stake_addr_info = cluster_obj.g_query.get_stake_addr_info(stake_addr.address) - if stake_addr_info: - LOGGER.debug(f"Re-registered stake address '{stake_addr.address}'") - break - else: - LOGGER.exception(f"Failed to re-register stake address '{stake_addr.address}'") - - return stake_addr_info + if not rereg_certs: + return None + return _CertificateAction( + address=stake_addr.address, + certificates=rereg_certs, + signing_keys=[stake_addr.skey_file], + deposit=rereg_deposit, + ) -def deregister_stake_addr( +def _stake_deregistration_action( *, cluster_obj: clusterlib.ClusterLib, - payment_addr: clusterlib.AddressRecord, stake_addr: clusterlib.AddressRecord, stake_addr_info: clusterlib.StakeAddrInfo, name_template: str, deposit_amt: int, -) -> None: - """Deregister stake address.""" +) -> _CertificateAction: + """Prepare a reward withdrawal and stake deregistration.""" withdrawals = [] if stake_addr_info.reward_account_balance: withdrawals = [ @@ -119,57 +174,35 @@ def deregister_stake_addr( deposit_amt=deposit_amt, stake_vkey_file=stake_addr.vkey_file, ) - tx_files_deregister = clusterlib.TxFiles( - certificate_files=[stake_addr_dereg_cert], - signing_key_files=[payment_addr.skey_file, stake_addr.skey_file], + return _CertificateAction( + address=stake_addr.address, + certificates=[stake_addr_dereg_cert], + signing_keys=[stake_addr.skey_file], + withdrawals=withdrawals, + deposit=-deposit_amt, ) - try: - cluster_obj.g_transaction.send_tx( - src_address=payment_addr.address, - tx_name=f"rf_{name_template}_dereg_withdraw_stake", - tx_files=tx_files_deregister, - withdrawals=withdrawals, - deposit=-deposit_amt, - ) - except clusterlib.CLIError: - LOGGER.exception(f"Failed to deregister stake address '{stake_addr.address}'") - else: - LOGGER.debug(f"Deregistered stake address '{stake_addr.address}'") - -def retire_drep( +def _drep_retirement_action( *, cluster_obj: clusterlib.ClusterLib, - payment_addr: clusterlib.AddressRecord, drep_keys: clusterlib.KeyPair, name_template: str, deposit_amt: int, -) -> None: - """Retire a DRep.""" +) -> _CertificateAction: + """Prepare a DRep retirement and deposit refund.""" ret_cert = cluster_obj.g_governance.drep.gen_retirement_cert( cert_name=f"rf_{name_template}", deposit_amt=deposit_amt, drep_vkey_file=drep_keys.vkey_file, ) - tx_files = clusterlib.TxFiles( - certificate_files=[ret_cert], - signing_key_files=[payment_addr.skey_file, drep_keys.skey_file], + return _CertificateAction( + address=name_template, + certificates=[ret_cert], + signing_keys=[drep_keys.skey_file], + deposit=-deposit_amt, ) - try: - cluster_obj.g_transaction.send_tx( - src_address=payment_addr.address, - tx_name=f"rf_{name_template}_retire_drep", - tx_files=tx_files, - deposit=-deposit_amt, - ) - except clusterlib.CLIError as excp: - if "ConwayDRepNotRegistered" not in str(excp): - LOGGER.exception(f"Failed to retire a DRep '{name_template}'") - else: - LOGGER.debug(f"Retired a DRep '{name_template}'") - def get_tx_inputs( *, cluster_obj: clusterlib.ClusterLib, src_addrs: list[clusterlib.AddressRecord] @@ -180,15 +213,19 @@ def get_tx_inputs( Don't exclude UTxOs with datum. All the UTxOs here has 'skey', and so the UTxOs with datum are spendable reference inputs. """ - recs = [] - for a in src_addrs: - utxos = cluster_obj.g_query.get_utxo(address=a.address) + recs: TxInputGroup = [] + for addresses in itertools.batched(src_addrs, QUERY_BATCH_SIZE, strict=False): + utxos = cluster_obj.g_query.get_utxo(address=[a.address for a in addresses]) utxos_ids_excluded = { f"{u.utxo_hash}#{u.utxo_ix}" for u in utxos if u.coin != clusterlib.DEFAULT_COIN } - txins_ok = [u for u in utxos if f"{u.utxo_hash}#{u.utxo_ix}" not in utxos_ids_excluded] - if txins_ok: - recs.append((txins_ok, a.skey_file)) + by_address: dict[str, list[clusterlib.UTXOData]] = {} + for utxo in utxos: + if f"{utxo.utxo_hash}#{utxo.utxo_ix}" not in utxos_ids_excluded: + by_address.setdefault(utxo.address, []).append(utxo) + recs.extend( + (by_address[a.address], a.skey_file) for a in addresses if a.address in by_address + ) return recs @@ -261,6 +298,7 @@ def return_funds_to_faucet( # Tx inputs deduplication is not strictly needed, as we are deduplicating the # address files. Keeping it here for separation of concerns. + submitted_inputs = [] batch_num = 1 for batch in batch_tx_inputs(tx_inputs=dedup_tx_inputs(tx_inputs=tx_inputs)): txins, skeys = flatten_tx_inputs(tx_inputs=batch) @@ -272,7 +310,7 @@ def return_funds_to_faucet( last_excp = None for attempt in range(1, 4): try: - cluster_obj.g_transaction.send_tx( + tx_output = cluster_obj.g_transaction.send_tx( src_address=txins[0].address, tx_name=f"{batch_tx_name}_try{attempt}", txins=txins, @@ -288,6 +326,7 @@ def return_funds_to_faucet( continue LOGGER.exception(f"Failed to return funds from addresses for '{batch_tx_name}'") else: + submitted_inputs.append(tx_output.txins[0]) LOGGER.debug(f"Returned funds from addresses '{batch_tx_name}'") break else: @@ -296,7 +335,12 @@ def return_funds_to_faucet( ) batch_num += 1 - cluster_obj.wait_for_new_block(new_blocks=3) + if submitted_inputs: + try: + clusterlib_utils.check_txins_spent(cluster_obj=cluster_obj, txins=submitted_inputs) + except AssertionError: + LOGGER.exception("Some return-funds transactions were not confirmed") + LOGGER.info(f"Returned funds: submitted {len(submitted_inputs)} transactions") def create_addr_record(*, addr_file: pl.Path) -> clusterlib.AddressRecord: @@ -357,89 +401,204 @@ def dedup_addresses( yield fpath -def cleanup_addresses( +def _fund_cleanup_addresses( *, cluster_obj: clusterlib.ClusterLib, - location: pl.Path, faucet_payment: clusterlib.AddressRecord, -) -> None: - """Cleanup addresses.""" - files_found = list(dedup_addresses(filter_addr_files(find_addr_files(location)))) - num_threads = min(10, (len(files_found) // 200) + 1) - file_chunks = [files_found[i::num_threads] for i in range(num_threads)] - - # Fund the addresses that will pay for fees + count: int, + name_template: str, +) -> list[clusterlib.AddressRecord]: + """Fund exactly the workers needed for certificate cleanup.""" + if not count: + return [] fund_addrs = [ - cluster_obj.g_address.gen_payment_addr_and_keys(name=f"addrs_cleanup{i}") - for i in range(num_threads) + cluster_obj.g_address.gen_payment_addr_and_keys(name=f"{name_template}{i}") + for i in range(count) ] - fund_dst = [clusterlib.TxOut(address=f.address, amount=300_000_000) for f in fund_addrs] - fund_tx_files = clusterlib.TxFiles(signing_key_files=[faucet_payment.skey_file]) cluster_obj.g_transaction.send_tx( src_address=faucet_payment.address, - tx_name="fund_addrs_cleanup", - txouts=fund_dst, - tx_files=fund_tx_files, + tx_name=f"fund_{name_template}", + txouts=[clusterlib.TxOut(address=f.address, amount=300_000_000) for f in fund_addrs], + tx_files=clusterlib.TxFiles(signing_key_files=[faucet_payment.skey_file]), ) + return fund_addrs - stake_deposit_amt = cluster_obj.g_query.get_address_deposit() - def _run(files: list[pl.Path], f_addr: clusterlib.AddressRecord) -> TxInputGroup: - tx_inputs = [] - for fpath in files: - f_name = fpath.name +def _query_registered_stake_addresses( + *, cluster_obj: clusterlib.ClusterLib, registered: list[_CertificateAction] +) -> dict[str, clusterlib.StakeAddrInfo]: + """Query newly registered accounts, retrying lagging visibility checks together.""" + pending = {a.address for a in registered} + stake_info = {} + for attempt in range(3): + if not pending: + break + if attempt: + cluster_obj.wait_for_new_block() + for address in sorted(pending): + info = cluster_obj.g_query.get_stake_addr_info(address) + if info and info.vote_delegation: + stake_info[address] = info + pending.remove(address) + return stake_info - # Add random sleep for < 0.5s to prevent - # "Network.Socket.connect: : resource exhausted" - time.sleep(random.random() / 2) - if f_name.endswith("_stake.addr"): - try: - stake_addr = create_addr_record(addr_file=fpath) - except ValueError as exc: - LOGGER.debug(f"Skipping: {exc}") - continue +def _cleanup_stake_addresses( + *, + cluster_obj: clusterlib.ClusterLib, + stake_addrs: list[clusterlib.AddressRecord], + payment_addr: clusterlib.AddressRecord, + deposit_amt: int, + name_template: str, +) -> None: + """Re-register, discover rewards, then withdraw and deregister in batches.""" + # Recycle registration deposits before starting another batch. Otherwise a large + # artifact directory could exhaust a worker's funding during re-registration. + if len(stake_addrs) > CERT_BATCH_SIZE: + for i, batch in enumerate(itertools.batched(stake_addrs, CERT_BATCH_SIZE, strict=False)): + _cleanup_stake_addresses( + cluster_obj=cluster_obj, + stake_addrs=list(batch), + payment_addr=payment_addr, + deposit_amt=deposit_amt, + name_template=f"{name_template}_part{i}", + ) + return + stake_info = {} + registrations = [] + for i, stake_addr in enumerate(stake_addrs): + # Limit concurrent node connections during discovery, not payment UTxO batching. + time.sleep(random.random() / 2) + info = cluster_obj.g_query.get_stake_addr_info(stake_addr.address) + stake_info[stake_addr.address] = info + action = _stake_registration_action( + cluster_obj=cluster_obj, + stake_addr=stake_addr, + stake_addr_info=info, + name_template=f"{name_template}_{i}", + deposit_amt=deposit_amt, + ) + if action: + registrations.append(action) - stake_addr_info = reregister_stake_addr( - cluster_obj=cluster_obj, - payment_addr=f_addr, - stake_addr=stake_addr, - stake_addr_info=cluster_obj.g_query.get_stake_addr_info(stake_addr.address), - name_template=f_name, - deposit_amt=stake_deposit_amt, - ) + registered = _submit_certificate_batches( + cluster_obj=cluster_obj, + payment_addr=payment_addr, + actions=registrations, + name_template=f"{name_template}_rereg", + ) + # Previously unregistered accounts may now expose rewards. + registered_info = _query_registered_stake_addresses( + cluster_obj=cluster_obj, registered=registered + ) + stake_info.update(registered_info) + needs_registration = {a.address for a in registrations} + deregistrations = [] + for i, stake_addr in enumerate(stake_addrs): + if stake_addr.address in needs_registration and stake_addr.address not in registered_info: + LOGGER.error(f"Skipping unconfirmed stake registration '{stake_addr.address}'") + continue + info = stake_info[stake_addr.address] + refund = info.registration_deposit if info.registration_deposit >= 0 else deposit_amt + deregistrations.append( + _stake_deregistration_action( + cluster_obj=cluster_obj, + stake_addr=stake_addr, + stake_addr_info=info, + name_template=f"{name_template}_{i}", + deposit_amt=refund, + ) + ) + _submit_certificate_batches( + cluster_obj=cluster_obj, + payment_addr=payment_addr, + actions=deregistrations, + name_template=f"{name_template}_dereg", + ) - deregister_stake_addr( + +def cleanup_addresses( + *, + cluster_obj: clusterlib.ClusterLib, + location: pl.Path, + faucet_payment: clusterlib.AddressRecord, +) -> None: + """Discover payment inputs and clean stake accounts using bounded batches.""" + started = time.monotonic() + files_found = list(dedup_addresses(filter_addr_files(find_addr_files(location)))) + payment_addrs = [] + stake_addrs = [] + for fpath in files_found: + try: + address = create_addr_record(addr_file=fpath) + except ValueError as exc: + LOGGER.debug(f"Skipping: {exc}") + continue + if fpath.name.endswith("_stake.addr"): + stake_addrs.append(address) + else: + payment_addrs.append(address) + LOGGER.info( + f"Address discovery: {len(payment_addrs)} payment, {len(stake_addrs)} stake accounts" + ) + tx_inputs = get_tx_inputs(cluster_obj=cluster_obj, src_addrs=payment_addrs) + LOGGER.info(f"Payment UTxO discovery finished in {time.monotonic() - started:.1f}s") + + num_threads = min(10, (len(stake_addrs) + CERT_BATCH_SIZE - 1) // CERT_BATCH_SIZE) + fund_addrs = _fund_cleanup_addresses( + cluster_obj=cluster_obj, + faucet_payment=faucet_payment, + count=num_threads, + name_template="addrs_cleanup", + ) + if stake_addrs: + stake_started = time.monotonic() + stake_deposit_amt = cluster_obj.g_query.get_address_deposit() + with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [ + executor.submit( + _cleanup_stake_addresses, cluster_obj=cluster_obj, - payment_addr=f_addr, - stake_addr=stake_addr, - stake_addr_info=stake_addr_info, - name_template=f_name, + stake_addrs=stake_addrs[i::num_threads], + payment_addr=addr, deposit_amt=stake_deposit_amt, + name_template=f"stake_cleanup{i}", ) - else: - try: - payment_addr = create_addr_record(addr_file=fpath) - except ValueError as exc: - LOGGER.debug(f"Skipping: {exc}") - continue - tx_inputs.extend(get_tx_inputs(cluster_obj=cluster_obj, src_addrs=[payment_addr])) - - return tx_inputs - - # Run cleanup / discovery in parallel - with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: - futures = [executor.submit(_run, ch, fund_addrs[i]) for i, ch in enumerate(file_chunks)] - results = [f.result() for f in futures] + for i, addr in enumerate(fund_addrs) + ] + for future in futures: + future.result() + LOGGER.info(f"Stake cleanup finished in {time.monotonic() - stake_started:.1f}s") - # Return funds from the collected UTxOs - tx_inputs = list(itertools.chain.from_iterable(results)) tx_inputs.extend(get_tx_inputs(cluster_obj=cluster_obj, src_addrs=fund_addrs)) return_funds_to_faucet( cluster_obj=cluster_obj, tx_inputs=tx_inputs, faucet_address=faucet_payment.address, ) + LOGGER.info(f"Address cleanup finished in {time.monotonic() - started:.1f}s") + + +def _find_drep_keys(location: pl.Path) -> list[clusterlib.KeyPair]: + """Find unique DRep keys, skipping missing or unreadable artifacts.""" + keys = [] + seen_keys = set() + for cert_file in find_cert_files(location): + vkey_file = cert_file.parent / cert_file.name.replace("_reg.cert", ".vkey") + skey_file = vkey_file.with_suffix(".skey") + if not (vkey_file.exists() and skey_file.exists()): + LOGGER.debug(f"Skipping DRep with missing keys: {cert_file}") + continue + try: + identity = json.loads(vkey_file.read_text())["cborHex"] + except (OSError, ValueError, KeyError): + LOGGER.exception(f"Skipping unreadable DRep key: {vkey_file}") + continue + if identity in seen_keys: + continue + seen_keys.add(identity) + keys.append(clusterlib.KeyPair(vkey_file=vkey_file, skey_file=skey_file)) + return keys def cleanup_certs( @@ -448,52 +607,61 @@ def cleanup_certs( location: pl.Path, faucet_payment: clusterlib.AddressRecord, ) -> None: - """Cleanup certificates.""" - files_found = list(find_cert_files(location)) - num_threads = min(10, (len(files_found) // 10) + 1) - file_chunks = [files_found[i::num_threads] for i in range(num_threads)] + """Retire registered DReps in batches, skipping duplicates and missing keys.""" + started = time.monotonic() + keys = _find_drep_keys(location) + if not keys: + LOGGER.info("No DRep keys to clean up") + return drep_deposit_amt = cluster_obj.g_query.get_drep_deposit() - # Fund the addresses that will pay for fees - fund_addrs = [ - cluster_obj.g_address.gen_payment_addr_and_keys(name=f"certs_cleanup{i}") - for i in range(num_threads + 1) - ] - fund_dst = [clusterlib.TxOut(address=f.address, amount=300_000_000) for f in fund_addrs] - fund_tx_files = clusterlib.TxFiles(signing_key_files=[faucet_payment.skey_file]) - cluster_obj.g_transaction.send_tx( - src_address=faucet_payment.address, - tx_name="fund_certs_cleanup", - txouts=fund_dst, - tx_files=fund_tx_files, - ) - - def _run(files: list[pl.Path], payment_addr: clusterlib.AddressRecord) -> None: - for cert_file in files: - # Add random sleep for < 0.5s to prevent - # "Network.Socket.connect: : resource exhausted" - time.sleep(random.random() / 2) + def _discover(item: tuple[int, clusterlib.KeyPair]) -> _CertificateAction | None: + i, drep_keys = item + time.sleep(random.random() / 2) + try: + state = cluster_obj.g_query.get_drep_state(drep_vkey_file=drep_keys.vkey_file) + except clusterlib.CLIError: + LOGGER.exception(f"Failed to query DRep: {drep_keys.vkey_file}") + return None + if not state: + return None + return _drep_retirement_action( + cluster_obj=cluster_obj, + drep_keys=drep_keys, + name_template=f"drep_cleanup{i}", + deposit_amt=state[0][1].get("deposit", drep_deposit_amt), + ) - f_name = cert_file.name - f_dir = cert_file.parent - vkey_file = f_dir / cert_file.name.replace("_reg.cert", ".vkey") - skey_file = vkey_file.with_suffix(".skey") - drep_keys = clusterlib.KeyPair(vkey_file=vkey_file, skey_file=skey_file) + with concurrent.futures.ThreadPoolExecutor(max_workers=min(10, len(keys))) as executor: + actions = [a for a in executor.map(_discover, enumerate(keys)) if a is not None] + LOGGER.info( + f"DRep discovery: {len(actions)}/{len(keys)} registered, " + f"finished in {time.monotonic() - started:.1f}s" + ) + if not actions: + return - retire_drep( + num_threads = min(10, (len(actions) + CERT_BATCH_SIZE - 1) // CERT_BATCH_SIZE) + fund_addrs = _fund_cleanup_addresses( + cluster_obj=cluster_obj, + faucet_payment=faucet_payment, + count=num_threads, + name_template="certs_cleanup", + ) + with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: + futures = [ + executor.submit( + _submit_certificate_batches, cluster_obj=cluster_obj, - payment_addr=payment_addr, - drep_keys=drep_keys, - name_template=f_name, - deposit_amt=drep_deposit_amt, + payment_addr=addr, + actions=actions[i::num_threads], + name_template=f"drep_cleanup{i}_retire", ) - # We don't need to deregister pools, because all tests clean the pools they created - - # Run cleanup in parallel - with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor: - futures = [executor.submit(_run, f, fund_addrs[i]) for i, f in enumerate(file_chunks)] - concurrent.futures.wait(futures) + for i, addr in enumerate(fund_addrs) + ] + for future in futures: + future.result() # Return funds from the addresses that paid for fees tx_inputs = get_tx_inputs(cluster_obj=cluster_obj, src_addrs=fund_addrs) @@ -503,6 +671,7 @@ def _run(files: list[pl.Path], payment_addr: clusterlib.AddressRecord) -> None: faucet_address=faucet_payment.address, tx_name="certs_cleanup_return", ) + LOGGER.info(f"Certificate cleanup finished in {time.monotonic() - started:.1f}s") def is_framework_testnet() -> bool: @@ -607,6 +776,7 @@ def cleanup( faucet_skey_file: clusterlib.FileType = "", ) -> None: """Cleanup a testnet with the help of testing artifacts.""" + started = time.monotonic() location = pl.Path(location).expanduser().resolve() faucet_payment = _get_faucet_payment_rec(address=faucet_address, skey_file=faucet_skey_file) @@ -617,9 +787,12 @@ def cleanup( cleanup_certs(cluster_obj=cluster_obj, location=location, faucet_payment=faucet_payment) # Defragment faucet address UTxOs + defrag_started = time.monotonic() defragment_utxos.defragment( cluster_obj=cluster_obj, address=faucet_payment.address, skey_file=faucet_payment.skey_file ) + LOGGER.info(f"Faucet defragmentation finished in {time.monotonic() - defrag_started:.1f}s") final_faucet_balance = cluster_obj.g_query.get_address_balance(address=faucet_payment.address) LOGGER.info(f"Final faucet balance: {final_faucet_balance}") + LOGGER.info(f"Cleanup finished in {time.monotonic() - started:.1f}s") diff --git a/framework_tests/test_testnet_cleanup.py b/framework_tests/test_testnet_cleanup.py new file mode 100644 index 000000000..eceaf4c86 --- /dev/null +++ b/framework_tests/test_testnet_cleanup.py @@ -0,0 +1,465 @@ +"""Unit tests for batched testnet cleanup without a running node.""" + +import dataclasses +import json +import pathlib as pl +from unittest import mock + +import pytest +from cardano_clusterlib import clusterlib + +from cardano_node_tests.utils import testnet_cleanup as cleanup + + +def _address(name: str) -> clusterlib.AddressRecord: + """Create a payment or stake record without writing key files.""" + return clusterlib.AddressRecord( + address=name, vkey_file=pl.Path(f"{name}.vkey"), skey_file=pl.Path(f"{name}.skey") + ) + + +def _info( + address: str = "", *, rewards: int = 0, delegation: str = "", deposit: int = 2_000_000 +) -> clusterlib.StakeAddrInfo: + """Create queried stake-account information, including unregistered accounts.""" + return clusterlib.StakeAddrInfo( + address=address, + delegation="", + reward_account_balance=rewards, + registration_deposit=deposit, + vote_delegation=delegation, + ) + + +def _action(name: str, *, deposit: int = 0) -> cleanup._CertificateAction: + """Create one certificate action for submission tests.""" + return cleanup._CertificateAction( + address=name, + certificates=[pl.Path(f"{name}.cert")], + signing_keys=[pl.Path(f"{name}.skey")], + deposit=deposit, + ) + + +@pytest.fixture +def cluster(monkeypatch: pytest.MonkeyPatch) -> mock.MagicMock: + """Provide a cluster mock with real scalar values and no discovery sleeps.""" + monkeypatch.setattr(cleanup.time, "sleep", lambda _: None) + obj = mock.MagicMock() + obj.g_query.get_address_deposit.return_value = 2_000_000 + obj.g_query.get_drep_deposit.return_value = 500_000_000 + obj.g_query.get_utxo.return_value = [] + obj.g_address.gen_payment_addr_and_keys.side_effect = _address + obj.g_stake_address.gen_stake_addr_registration_cert.return_value = pl.Path("register.cert") + obj.g_stake_address.gen_vote_delegation_cert.return_value = pl.Path("delegate.cert") + obj.g_stake_address.gen_stake_addr_deregistration_cert.return_value = pl.Path("deregister.cert") + obj.g_governance.drep.gen_retirement_cert.return_value = pl.Path("retire.cert") + return obj + + +def test_certificate_batch_balancing(cluster: mock.MagicMock): + """Aggregate certificates, deposits and withdrawals, deduplicating witnesses.""" + actions = [_action("stake0", deposit=-2_000_000), _action("stake1", deposit=-3_000_000)] + actions[0].withdrawals = [clusterlib.TxOut(address="stake0", amount=123)] + actions[1].signing_keys.append(actions[0].signing_keys[0]) + result = cleanup._submit_certificate_batches( + cluster_obj=cluster, payment_addr=_address("payer"), actions=actions, name_template="test" + ) + assert result == actions + cluster.g_transaction.send_tx.assert_called_once() + kwargs = cluster.g_transaction.send_tx.call_args.kwargs + assert kwargs["deposit"] == -5_000_000 + assert kwargs["withdrawals"] == actions[0].withdrawals + assert kwargs["tx_files"].certificate_files == [pl.Path("stake0.cert"), pl.Path("stake1.cert")] + assert kwargs["tx_files"].signing_key_files == [ + pl.Path("payer.skey"), + pl.Path("stake0.skey"), + pl.Path("stake1.skey"), + ] + assert kwargs.get("verify_tx", True) + + +def test_certificate_batch_limit(cluster: mock.MagicMock): + """Bound certificate batches and use distinct transaction artifact names.""" + actions = [_action(f"stake{i}") for i in range(cleanup.CERT_BATCH_SIZE + 1)] + assert ( + cleanup._submit_certificate_batches( + cluster_obj=cluster, + payment_addr=_address("payer"), + actions=actions, + name_template="test", + ) + == actions + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert [len(c.kwargs["tx_files"].certificate_files) for c in calls] == [20, 1] + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) + + +def test_rejected_certificate_isolation(cluster: mock.MagicMock): + """A stale credential does not prevent the other certificates being submitted.""" + actions = [_action("good"), _action("stale")] + + def send(**kwargs: object) -> None: + files = kwargs["tx_files"] + assert isinstance(files, clusterlib.TxFiles) + if pl.Path("stale.cert") in files.certificate_files: + msg = "ConwayDRepNotRegistered" + raise clusterlib.CLIError(msg) + + cluster.g_transaction.send_tx.side_effect = send + assert ( + cleanup._submit_certificate_batches( + cluster_obj=cluster, + payment_addr=_address("payer"), + actions=actions, + name_template="test", + ) + == actions[:1] + ) + assert cluster.g_transaction.send_tx.call_count == 3 + + +def test_oversized_certificate_batch(cluster: mock.MagicMock): + """Split an oversized transaction into smaller confirmed batches.""" + actions = [_action("a"), _action("b")] + cluster.g_transaction.send_tx.side_effect = [clusterlib.CLIError("MaxTxSizeUTxO"), None, None] + assert ( + cleanup._submit_certificate_batches( + cluster_obj=cluster, + payment_addr=_address("payer"), + actions=actions, + name_template="test", + ) + == actions + ) + + +@pytest.mark.parametrize("error", ["didn't make it to the chain", "Failed to resubmit"]) +def test_uncertain_submission_not_split(cluster: mock.MagicMock, error: str): + """Do not retry parts of a transaction whose confirmation is uncertain.""" + cluster.g_transaction.send_tx.side_effect = clusterlib.CLIError(error) + assert not cleanup._submit_certificate_batches( + cluster_obj=cluster, + payment_addr=_address("payer"), + actions=[_action("a"), _action("b")], + name_template="test", + ) + cluster.g_transaction.send_tx.assert_called_once() + + +def test_unregistered_accounts_recover_rewards(cluster: mock.MagicMock): + """Preserve re-registration and query newly visible rewards before deregistration.""" + cluster.g_query.get_stake_addr_info.side_effect = [ + _info(), + _info(), + _info("stake0", rewards=123, delegation="alwaysAbstain"), + _info("stake1", rewards=456, delegation="alwaysAbstain"), + ] + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0"), _address("stake1")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert len(calls) == 2 + assert calls[0].kwargs["deposit"] == 4_000_000 + assert len(calls[0].kwargs["tx_files"].certificate_files) == 4 + assert calls[1].kwargs["deposit"] == -4_000_000 + assert calls[1].kwargs["withdrawals"] == [ + clusterlib.TxOut(address="stake0", amount=123), + clusterlib.TxOut(address="stake1", amount=456), + ] + cluster.wait_for_new_block.assert_not_called() + + +def test_registered_account_delegation(cluster: mock.MagicMock): + """Delegate votes only for registered accounts with rewards and no delegation.""" + cluster.g_query.get_stake_addr_info.side_effect = [ + _info("stake0", rewards=123), + _info("stake0", rewards=123, delegation="alwaysAbstain", deposit=1_000_000), + ] + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + cluster.g_stake_address.gen_stake_addr_registration_cert.assert_not_called() + cluster.g_stake_address.gen_vote_delegation_cert.assert_called_once() + calls = cluster.g_transaction.send_tx.call_args_list + assert [c.kwargs["deposit"] for c in calls] == [0, -1_000_000] + + +def test_registered_zero_reward_account(cluster: mock.MagicMock): + """Deregister a registered account directly when no vote delegation is needed.""" + cluster.g_query.get_stake_addr_info.return_value = _info("stake0") + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + cluster.g_stake_address.gen_vote_delegation_cert.assert_not_called() + cluster.g_transaction.send_tx.assert_called_once() + assert cluster.g_transaction.send_tx.call_args.kwargs["withdrawals"] == [] + + +def test_failed_registration_not_deregistered(cluster: mock.MagicMock): + """Do not build deregistration certificates after rejected re-registration.""" + cluster.g_query.get_stake_addr_info.return_value = _info() + cluster.g_transaction.send_tx.side_effect = clusterlib.CLIError("rejected") + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + cluster.g_stake_address.gen_stake_addr_deregistration_cert.assert_not_called() + cluster.g_transaction.send_tx.assert_called_once() + + +def test_registration_visibility_retries_together(cluster: mock.MagicMock): + """Wait for lagging registration queries once per round, not once per account.""" + cluster.g_query.get_stake_addr_info.side_effect = [ + _info(), + _info(), + _info(), + _info(), + _info("stake0", delegation="alwaysAbstain"), + _info("stake1", delegation="alwaysAbstain"), + ] + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0"), _address("stake1")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + cluster.wait_for_new_block.assert_called_once() + assert cluster.g_transaction.send_tx.call_count == 2 + + +def test_unqueryable_registration_not_deregistered(cluster: mock.MagicMock): + """Skip an account that remains unqueryable after confirmed registration.""" + cluster.g_query.get_stake_addr_info.return_value = _info() + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=[_address("stake0")], + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + cluster.g_stake_address.gen_stake_addr_deregistration_cert.assert_not_called() + assert cluster.wait_for_new_block.call_count == 2 + + +def test_stake_batches_recycle_deposits(cluster: mock.MagicMock): + """Complete deregistration before re-registering the next funded batch.""" + addresses = [_address(f"stake{i}") for i in range(cleanup.CERT_BATCH_SIZE + 1)] + queried = set() + + def info(address: str) -> clusterlib.StakeAddrInfo: + if address in queried: + return _info(address, delegation="alwaysAbstain") + queried.add(address) + return _info() + + cluster.g_query.get_stake_addr_info.side_effect = info + cleanup._cleanup_stake_addresses( + cluster_obj=cluster, + stake_addrs=addresses, + payment_addr=_address("payer"), + deposit_amt=2_000_000, + name_template="test", + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert [c.kwargs["deposit"] for c in calls] == [40_000_000, -40_000_000, 2_000_000, -2_000_000] + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) + + +def test_utxo_queries_batched(cluster: mock.MagicMock): + """Query bounded address groups and associate each input with its signing key.""" + addresses = [_address(f"payment{i}") for i in range(cleanup.QUERY_BATCH_SIZE + 1)] + first = clusterlib.UTXOData(utxo_hash="tx", utxo_ix=0, amount=10, address=addresses[0].address) + last = dataclasses.replace(first, utxo_ix=1, address=addresses[-1].address) + cluster.g_query.get_utxo.side_effect = [[first], [last]] + assert cleanup.get_tx_inputs(cluster_obj=cluster, src_addrs=addresses) == [ + ([first], addresses[0].skey_file), + ([last], addresses[-1].skey_file), + ] + assert cluster.g_query.get_utxo.call_args_list == [ + mock.call(address=[a.address for a in addresses[:-1]]), + mock.call(address=[addresses[-1].address]), + ] + + +def test_utxo_tokens_excluded_datums_preserved(cluster: mock.MagicMock): + """Exclude entire multi-asset inputs, preserving spendable key inputs with datum.""" + address = _address("payment") + ada = clusterlib.UTXOData(utxo_hash="tx", utxo_ix=0, amount=10, address=address.address) + token = dataclasses.replace(ada, coin="policy.asset", amount=1) + datum = dataclasses.replace(ada, utxo_ix=1, datum_hash="datum") + cluster.g_query.get_utxo.return_value = [ada, token, datum] + assert cleanup.get_tx_inputs(cluster_obj=cluster, src_addrs=[address]) == [ + ([datum], address.skey_file) + ] + + +def test_empty_cleanup_no_funding(cluster: mock.MagicMock, tmp_path: pl.Path): + """Empty artifact directories cause no funding, certificate or return transactions.""" + cleanup.cleanup_addresses(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cluster.g_address.gen_payment_addr_and_keys.assert_not_called() + cluster.g_transaction.send_tx.assert_not_called() + cluster.g_query.get_utxo.assert_not_called() + cluster.wait_for_new_block.assert_not_called() + + +def test_payment_only_cleanup_no_funding(cluster: mock.MagicMock, tmp_path: pl.Path): + """Payment-only discovery needs no fee-paying helper addresses.""" + for suffix, content in [("addr", "payment"), ("vkey", "key"), ("skey", "key")]: + (tmp_path / f"payment.{suffix}").write_text(content) + cleanup.cleanup_addresses(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cluster.g_address.gen_payment_addr_and_keys.assert_not_called() + cluster.g_query.get_utxo.assert_called_once_with(address=["payment"]) + cluster.g_transaction.send_tx.assert_not_called() + + +def _write_drep(directory: pl.Path, name: str, identity: str) -> pl.Path: + """Write the artifacts used to locate and deduplicate DRep keys.""" + (directory / f"{name}_drep_reg.cert").write_text("certificate") + vkey = directory / f"{name}_drep.vkey" + vkey.write_text(json.dumps({"cborHex": identity})) + vkey.with_suffix(".skey").write_text("key") + return vkey + + +def test_drep_cleanup_filters_and_batches(cluster: mock.MagicMock, tmp_path: pl.Path): + """Deduplicate keys, skip retired DReps and batch actual deposit refunds.""" + _write_drep(tmp_path, "active0", "00") + _write_drep(tmp_path, "duplicate", "00") + _write_drep(tmp_path, "active1", "01") + _write_drep(tmp_path, "retired", "02") + (tmp_path / "missing_drep_reg.cert").write_text("certificate") + cluster.g_query.get_drep_state.side_effect = lambda drep_vkey_file: ( + [[{}, {"deposit": 123}]] + if json.loads(drep_vkey_file.read_text())["cborHex"] in {"00", "01"} + else [] + ) + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + assert cluster.g_query.get_drep_state.call_count == 3 + cluster.g_address.gen_payment_addr_and_keys.assert_called_once_with(name="certs_cleanup0") + calls = cluster.g_transaction.send_tx.call_args_list + assert len(calls) == 2 + assert len(calls[0].kwargs["txouts"]) == 1 + assert calls[1].kwargs["deposit"] == -246 + assert len(calls[1].kwargs["tx_files"].certificate_files) == 2 + + +def test_retired_dreps_no_funding(cluster: mock.MagicMock, tmp_path: pl.Path): + """Do not fund helpers when all discovered DReps are already retired.""" + _write_drep(tmp_path, "retired", "00") + cluster.g_query.get_drep_state.return_value = [] + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cluster.g_address.gen_payment_addr_and_keys.assert_not_called() + cluster.g_transaction.send_tx.assert_not_called() + + +@pytest.mark.parametrize("content", ["not json", "{}"]) +def test_unreadable_drep_keys_skipped(cluster: mock.MagicMock, tmp_path: pl.Path, content: str): + """Malformed key artifacts do not prevent cleanup of other credentials.""" + vkey = _write_drep(tmp_path, "invalid", "00") + vkey.write_text(content) + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cluster.g_query.get_drep_state.assert_not_called() + cluster.g_transaction.send_tx.assert_not_called() + + +def test_invalid_drep_query_skipped(cluster: mock.MagicMock, tmp_path: pl.Path): + """A key rejected by the CLI does not require funding a retirement worker.""" + _write_drep(tmp_path, "invalid", "00") + cluster.g_query.get_drep_state.side_effect = clusterlib.CLIError("invalid verification key") + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + cluster.g_transaction.send_tx.assert_not_called() + + +def test_drep_worker_exceptions_propagate(cluster: mock.MagicMock, tmp_path: pl.Path): + """Surface unexpected worker failures instead of silently ignoring futures.""" + _write_drep(tmp_path, "active", "00") + cluster.g_query.get_drep_state.side_effect = RuntimeError("query failed") + with pytest.raises(RuntimeError, match="query failed"): + cleanup.cleanup_certs(cluster_obj=cluster, location=tmp_path, faucet_payment=_address("f")) + + +def test_return_funds_shared_verification(cluster: mock.MagicMock, monkeypatch: pytest.MonkeyPatch): + """Verify disjoint return transactions together instead of waiting fixed blocks.""" + inputs = [ + clusterlib.UTXOData(utxo_hash="tx", utxo_ix=i, amount=10_000_000, address="payment") + for i in range(101) + ] + cluster.g_transaction.send_tx.side_effect = [ + mock.Mock(txins=inputs[:100]), + mock.Mock(txins=inputs[100:]), + ] + check = mock.Mock() + monkeypatch.setattr(cleanup.clusterlib_utils, "check_txins_spent", check) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[(inputs, pl.Path("payment.skey"))], faucet_address="f" + ) + check.assert_called_once_with(cluster_obj=cluster, txins=[inputs[0], inputs[100]]) + assert all(not c.kwargs["verify_tx"] for c in cluster.g_transaction.send_tx.call_args_list) + cluster.wait_for_new_block.assert_not_called() + + +def test_return_funds_fee_retry(cluster: mock.MagicMock, monkeypatch: pytest.MonkeyPatch): + """Preserve fee retries and verify only successfully submitted transactions.""" + utxo = clusterlib.UTXOData(utxo_hash="tx", utxo_ix=0, amount=10_000_000, address="payment") + cluster.g_transaction.send_tx.side_effect = [ + clusterlib.CLIError("FeeTooSmallUTxO"), + mock.Mock(txins=[utxo]), + ] + check = mock.Mock() + monkeypatch.setattr(cleanup.clusterlib_utils, "check_txins_spent", check) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[([utxo], pl.Path("payment.skey"))], faucet_address="f" + ) + assert [ + c.kwargs["witness_count_add"] for c in cluster.g_transaction.send_tx.call_args_list + ] == [2, 7] + check.assert_called_once_with(cluster_obj=cluster, txins=[utxo]) + + +def test_failed_returns_no_wait(cluster: mock.MagicMock, monkeypatch: pytest.MonkeyPatch): + """Do not wait for transactions that were rejected before submission.""" + utxo = clusterlib.UTXOData(utxo_hash="tx", utxo_ix=0, amount=10_000_000, address="payment") + cluster.g_transaction.send_tx.side_effect = clusterlib.CLIError("rejected") + check = mock.Mock() + monkeypatch.setattr(cleanup.clusterlib_utils, "check_txins_spent", check) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[([utxo], pl.Path("payment.skey"))], faucet_address="f" + ) + check.assert_not_called() + cluster.wait_for_new_block.assert_not_called() + + +def test_unconfirmed_returns_logged( + cluster: mock.MagicMock, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +): + """Preserve best-effort cleanup when collective return verification fails.""" + utxo = clusterlib.UTXOData(utxo_hash="tx", utxo_ix=0, amount=10_000_000, address="payment") + cluster.g_transaction.send_tx.return_value = mock.Mock(txins=[utxo]) + monkeypatch.setattr( + cleanup.clusterlib_utils, + "check_txins_spent", + mock.Mock(side_effect=AssertionError("unspent")), + ) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[([utxo], pl.Path("payment.skey"))], faucet_address="f" + ) + assert "Some return-funds transactions were not confirmed" in caplog.text From ebb3551c0296b997fd5077b48ffd479169032536 Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 6 Nov 2026 18:52:48 +0100 Subject: [PATCH 2/4] fix: exclude faucet from cleanup discovery Exclude the supplied faucet by address regardless of its artifact filename. This prevents cached faucet inputs from becoming stale when worker funding spends them before the return transaction. Add regression coverage for an alternate faucet filename with and without stake cleanup, preserving unrelated and worker inputs. --- cardano_node_tests/utils/testnet_cleanup.py | 4 ++ framework_tests/test_testnet_cleanup.py | 50 +++++++++++++++++++++ 2 files changed, 54 insertions(+) diff --git a/cardano_node_tests/utils/testnet_cleanup.py b/cardano_node_tests/utils/testnet_cleanup.py index ab2e631f1..04b2a2e9c 100644 --- a/cardano_node_tests/utils/testnet_cleanup.py +++ b/cardano_node_tests/utils/testnet_cleanup.py @@ -534,6 +534,10 @@ def cleanup_addresses( except ValueError as exc: LOGGER.debug(f"Skipping: {exc}") continue + # The supplied faucet may appear under any artifact filename. Its inputs + # must not be cached here because worker funding spends them below. + if address.address == faucet_payment.address: + continue if fpath.name.endswith("_stake.addr"): stake_addrs.append(address) else: diff --git a/framework_tests/test_testnet_cleanup.py b/framework_tests/test_testnet_cleanup.py index eceaf4c86..bb1cc0d5a 100644 --- a/framework_tests/test_testnet_cleanup.py +++ b/framework_tests/test_testnet_cleanup.py @@ -331,6 +331,56 @@ def test_payment_only_cleanup_no_funding(cluster: mock.MagicMock, tmp_path: pl.P cluster.g_transaction.send_tx.assert_not_called() +@pytest.mark.parametrize("with_stake", [False, True]) +def test_actual_faucet_excluded_from_discovery( + cluster: mock.MagicMock, + tmp_path: pl.Path, + monkeypatch: pytest.MonkeyPatch, + with_stake: bool, +): + """Exclude a renamed faucet before caching inputs that worker funding spends.""" + artifacts = [("funding-wallet", "faucet_address"), ("payment", "payment")] + if with_stake: + artifacts.append(("account_stake", "stake")) + for name, address in artifacts: + for suffix, content in [("addr", address), ("vkey", "key"), ("skey", "key")]: + (tmp_path / f"{name}.{suffix}").write_text(content) + + payment_input = clusterlib.UTXOData( + utxo_hash="payment_tx", utxo_ix=0, amount=10_000_000, address="payment" + ) + worker_input = dataclasses.replace( + payment_input, utxo_hash="funding_tx", address="addrs_cleanup0" + ) + faucet_input = dataclasses.replace( + payment_input, utxo_hash="faucet_tx", address="faucet_address" + ) + available = {u.address: [u] for u in [payment_input, worker_input, faucet_input]} + cluster.g_query.get_utxo.side_effect = lambda address: [ + u for a in address for u in available[a] + ] + monkeypatch.setattr(cleanup, "_cleanup_stake_addresses", mock.Mock()) + return_funds = mock.Mock() + monkeypatch.setattr(cleanup, "return_funds_to_faucet", return_funds) + + cleanup.cleanup_addresses( + cluster_obj=cluster, location=tmp_path, faucet_payment=_address("faucet_address") + ) + + expected_inputs = [([payment_input], tmp_path / "payment.skey")] + if with_stake: + expected_inputs.append(([worker_input], pl.Path("addrs_cleanup0.skey"))) + cluster.g_transaction.send_tx.assert_called_once() + else: + cluster.g_transaction.send_tx.assert_not_called() + return_funds.assert_called_once_with( + cluster_obj=cluster, tx_inputs=expected_inputs, faucet_address="faucet_address" + ) + assert all( + "faucet_address" not in c.kwargs["address"] for c in cluster.g_query.get_utxo.call_args_list + ) + + def _write_drep(directory: pl.Path, name: str, identity: str) -> pl.Path: """Write the artifacts used to locate and deduplicate DRep keys.""" (directory / f"{name}_drep_reg.cert").write_text("certificate") From 5714379e6c3ccc267c0d13517f479339b67844a3 Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 6 Nov 2026 19:11:15 +0100 Subject: [PATCH 3/4] fix: isolate rejected payment cleanup groups Split definitely rejected payment batches by address while keeping each address's inputs together. Refresh only original UTxOs, preserve fee retries and artifact uniqueness, and confirm submissions together. Quarantine uncertain inputs instead of rebuilding or reusing them. Treat unknown, infrastructure and ambiguous spent-input errors conservatively without recursive retries. Add mocked coverage for isolation, stale inputs, retry bounds and uncertain outcomes. Validation: 54 targeted tests and 605 framework tests passed, with one expected failure. Lint passed. --- cardano_node_tests/utils/testnet_cleanup.py | 213 +++++++++++++--- framework_tests/test_testnet_cleanup.py | 267 ++++++++++++++++++++ 2 files changed, 440 insertions(+), 40 deletions(-) diff --git a/cardano_node_tests/utils/testnet_cleanup.py b/cardano_node_tests/utils/testnet_cleanup.py index 04b2a2e9c..cc64a0bf9 100644 --- a/cardano_node_tests/utils/testnet_cleanup.py +++ b/cardano_node_tests/utils/testnet_cleanup.py @@ -14,6 +14,8 @@ import logging import pathlib as pl import random +import re +import subprocess import time import typing as tp @@ -282,6 +284,159 @@ def flatten_tx_inputs( return list(itertools.chain.from_iterable(utxo_lists)), list(skeys) +def _payment_rejection_reason(error: clusterlib.CLIError) -> str: + """Recognize only stateless ledger failures that cannot follow acceptance. + + `send_tx(verify_tx=False)` uses `submit_tx_bare`, which retries MempoolTxTooSlow; + `cli` also retries resource-exhausted/vanished connections. Consequently BadInputsUTxO + (including AllInputsAreSpent) is ambiguous: an earlier attempt may have succeeded. + Unknown, transport and timeout failures must never trigger rebuilding or splitting. + """ + messages = [] + current: BaseException | None = error + seen = set() + while current is not None and id(current) not in seen: + seen.add(id(current)) + if isinstance(current, (subprocess.TimeoutExpired, OSError)): + return "" + messages.append(str(current)) + current = current.__cause__ or current.__context__ + message = "\n".join(messages) + uncertain = ( + "MempoolTxTooSlow", + "timeout", + "timed out", + "resource exhausted", + "resource vanished", + "broken pipe", + "MuxError", + "Failed to submit", + "Failed to resubmit", + ) + if any(marker.lower() in message.lower() for marker in uncertain): + return "" + reasons = ( + "FeeTooSmallUTxO", + "MissingVKeyWitnessesUTXOW", + "InvalidWitnessesUTXOW", + "MaxTxSizeUTxO", + "OutputTooSmallUTxO", + "BabbageOutputTooSmallUTxO", + "OutputTooBigUTxO", + ) + return next((reason for reason in reasons if re.search(rf"\b{reason}\b", str(error))), "") + + +def _retain_payment_inputs(tx_inputs: TxInputGroup, keep: set[str]) -> TxInputGroup: + """Retain original inputs by ID, dropping empty groups and duplicate inputs.""" + retained = [] + seen = set() + for utxos, path in tx_inputs: + group = [] + for utxo in utxos: + identity = f"{utxo.utxo_hash}#{utxo.utxo_ix}" + if identity in keep and identity not in seen: + group.append(utxo) + seen.add(identity) + if group: + retained.append((group, path)) + return retained + + +def _send_payment_batch( + *, cluster_obj: clusterlib.ClusterLib, batch: TxInputGroup, faucet_address: str, tx_name: str +) -> clusterlib.TxRawOutput: + """Submit one payment batch with at most three fee-adjustment attempts.""" + txins, skeys = flatten_tx_inputs(tx_inputs=batch) + witness_count_add = max(2, len(skeys) // 20) + for attempt in range(1, 4): + try: + return cluster_obj.g_transaction.send_tx( + src_address=txins[0].address, + tx_name=f"{tx_name}_try{attempt}", + txins=txins, + txouts=[clusterlib.TxOut(address=faucet_address, amount=-1)], + tx_files=clusterlib.TxFiles(signing_key_files=skeys), + witness_count_add=witness_count_add, + verify_tx=False, + ) + except clusterlib.CLIError as exc: + if _payment_rejection_reason(exc) != "FeeTooSmallUTxO" or attempt == 3: + raise + witness_count_add += 5 + msg = "Unreachable payment retry state" + raise AssertionError(msg) # pragma: no cover + + +def _return_payment_batch( + *, + cluster_obj: clusterlib.ClusterLib, + batch: TxInputGroup, + faucet_address: str, + tx_name: str, + submitted_inputs: list[clusterlib.UTXOData], + unavailable: set[str], +) -> None: + """Isolate definite failures by address, quarantining unresolved submissions. + + Each child contains strictly fewer addresses than its rejected parent, bounding + recursion. Never split a single address's UTxOs or retry systemic/unknown failures. + """ + identities = {f"{u.utxo_hash}#{u.utxo_ix}" for utxos, _ in batch for u in utxos} + batch = _retain_payment_inputs(batch, identities - unavailable) + if not batch: + return + txins, _ = flatten_tx_inputs(tx_inputs=batch) + identities = {f"{u.utxo_hash}#{u.utxo_ix}" for u in txins} + addresses = list(dict.fromkeys(u.address for u in txins)) + try: + tx_output = _send_payment_batch( + cluster_obj=cluster_obj, batch=batch, faucet_address=faucet_address, tx_name=tx_name + ) + except (clusterlib.CLIError, subprocess.TimeoutExpired, OSError) as exc: + reason = _payment_rejection_reason(exc) if isinstance(exc, clusterlib.CLIError) else "" + if not reason or reason == "FeeTooSmallUTxO" or len(addresses) == 1: + # Reserve even unresolved inputs: querying them as unspent does not rule + # out acceptance in the mempool. Later batches must not reuse them. + unavailable.update(identities) + LOGGER.exception( + f"Skipping payment group '{tx_name}': addresses={addresses}, " + f"UTxOs={sorted(identities)}" + ) + return + LOGGER.warning(f"Isolating rejected payment group '{tx_name}': {exc}") + else: + unavailable.update(identities) + submitted_inputs.append(tx_output.txins[0]) + LOGGER.debug(f"Returned funds from addresses '{tx_name}'") + return + + # Query exactly this group's original inputs, not its addresses. Intersect with + # the original set even if the query returns unexpected/new outputs. + try: + unspent = cluster_obj.g_query.get_utxo(txin=sorted(identities)) + except (clusterlib.CLIError, subprocess.TimeoutExpired, OSError): + unavailable.update(identities) + LOGGER.exception(f"Failed to refresh rejected payment group '{tx_name}'") + return + keep = {f"{u.utxo_hash}#{u.utxo_ix}" for u in unspent} - unavailable + remaining = _retain_payment_inputs(batch, keep) + midpoint = len(addresses) // 2 + left_addresses = set(addresses[:midpoint]) + left_inputs = {f"{u.utxo_hash}#{u.utxo_ix}" for u in txins if u.address in left_addresses} + for suffix, group_ids in [("0", left_inputs), ("1", identities - left_inputs)]: + group = _retain_payment_inputs(remaining, group_ids) + if group: + _return_payment_batch( + cluster_obj=cluster_obj, + batch=group, + faucet_address=faucet_address, + tx_name=f"{tx_name}_{suffix}", + submitted_inputs=submitted_inputs, + unavailable=unavailable, + ) + + def return_funds_to_faucet( *, cluster_obj: clusterlib.ClusterLib, @@ -289,51 +444,29 @@ def return_funds_to_faucet( faucet_address: str, tx_name: str = "", ) -> None: - """Send funds from `tx_inputs` to `faucet_address`.""" + """Return funds in batches, isolating definitely rejected groups by address. + + Confirm successful submissions collectively. Quarantine uncertain inputs for this + invocation rather than rebuilding transactions that may already be accepted. + """ tx_name = tx_name or helpers.get_timestamped_rand_str() tx_name = f"rf_{tx_name}" - # The amount of "-1" means all available funds. - fund_dst = [clusterlib.TxOut(address=faucet_address, amount=-1)] - # Tx inputs deduplication is not strictly needed, as we are deduplicating the # address files. Keeping it here for separation of concerns. - submitted_inputs = [] - batch_num = 1 - for batch in batch_tx_inputs(tx_inputs=dedup_tx_inputs(tx_inputs=tx_inputs)): - txins, skeys = flatten_tx_inputs(tx_inputs=batch) - fund_tx_files = clusterlib.TxFiles(signing_key_files=skeys) - batch_tx_name = f"{tx_name}_batch{batch_num}" - witness_count_add = max(2, len(skeys) // 20) - - # Try to return funds - last_excp = None - for attempt in range(1, 4): - try: - tx_output = cluster_obj.g_transaction.send_tx( - src_address=txins[0].address, - tx_name=f"{batch_tx_name}_try{attempt}", - txins=txins, - txouts=fund_dst, - tx_files=fund_tx_files, - witness_count_add=witness_count_add, - verify_tx=False, - ) - except clusterlib.CLIError as excp: - last_excp = excp - if "FeeTooSmallUTxO" in str(excp): - witness_count_add += 5 - continue - LOGGER.exception(f"Failed to return funds from addresses for '{batch_tx_name}'") - else: - submitted_inputs.append(tx_output.txins[0]) - LOGGER.debug(f"Returned funds from addresses '{batch_tx_name}'") - break - else: - LOGGER.error( - f"Failed to return funds from addresses for '{batch_tx_name}'", exc_info=last_excp - ) - batch_num += 1 + submitted_inputs: list[clusterlib.UTXOData] = [] + unavailable: set[str] = set() + for batch_num, batch in enumerate( + batch_tx_inputs(tx_inputs=dedup_tx_inputs(tx_inputs=tx_inputs)), start=1 + ): + _return_payment_batch( + cluster_obj=cluster_obj, + batch=batch, + faucet_address=faucet_address, + tx_name=f"{tx_name}_batch{batch_num}", + submitted_inputs=submitted_inputs, + unavailable=unavailable, + ) if submitted_inputs: try: diff --git a/framework_tests/test_testnet_cleanup.py b/framework_tests/test_testnet_cleanup.py index bb1cc0d5a..b53b6f010 100644 --- a/framework_tests/test_testnet_cleanup.py +++ b/framework_tests/test_testnet_cleanup.py @@ -3,6 +3,8 @@ import dataclasses import json import pathlib as pl +import subprocess +import typing as tp from unittest import mock import pytest @@ -463,6 +465,10 @@ def test_return_funds_shared_verification(cluster: mock.MagicMock, monkeypatch: cluster_obj=cluster, tx_inputs=[(inputs, pl.Path("payment.skey"))], faucet_address="f" ) check.assert_called_once_with(cluster_obj=cluster, txins=[inputs[0], inputs[100]]) + assert [len(c.kwargs["txins"]) for c in cluster.g_transaction.send_tx.call_args_list] == [ + 100, + 1, + ] assert all(not c.kwargs["verify_tx"] for c in cluster.g_transaction.send_tx.call_args_list) cluster.wait_for_new_block.assert_not_called() @@ -483,6 +489,8 @@ def test_return_funds_fee_retry(cluster: mock.MagicMock, monkeypatch: pytest.Mon c.kwargs["witness_count_add"] for c in cluster.g_transaction.send_tx.call_args_list ] == [2, 7] check.assert_called_once_with(cluster_obj=cluster, txins=[utxo]) + calls = cluster.g_transaction.send_tx.call_args_list + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) def test_failed_returns_no_wait(cluster: mock.MagicMock, monkeypatch: pytest.MonkeyPatch): @@ -513,3 +521,262 @@ def test_unconfirmed_returns_logged( cluster_obj=cluster, tx_inputs=[([utxo], pl.Path("payment.skey"))], faucet_address="f" ) assert "Some return-funds transactions were not confirmed" in caplog.text + + +def _payment_inputs(names: list[str], *, per_address: int = 2) -> cleanup.TxInputGroup: + """Create multiple original UTxOs per payment address.""" + return [ + ( + [ + clusterlib.UTXOData( + utxo_hash=f"tx_{name}", utxo_ix=i, amount=10_000_000, address=name + ) + for i in range(per_address) + ], + pl.Path(f"{name}.skey"), + ) + for name in names + ] + + +def _payment_sender(bad: set[str]) -> tp.Callable[..., mock.Mock]: + """Reject any transaction containing a specified address, without submitting it.""" + + def send(*, txins: list[clusterlib.UTXOData], **_kwargs: tp.Any) -> mock.Mock: + if any(u.address in bad for u in txins): + msg = "ShelleyTxValidationError (MissingVKeyWitnessesUTXOW bad_key)" + raise clusterlib.CLIError(msg) + return mock.Mock(txins=txins) + + return send + + +@pytest.fixture +def payment_check(monkeypatch: pytest.MonkeyPatch) -> mock.Mock: + """Record collective confirmation without making node queries or waiting.""" + check = mock.Mock() + monkeypatch.setattr(cleanup.clusterlib_utils, "check_txins_spent", check) + return check + + +def test_payment_rejection_recovers_unaffected( + cluster: mock.MagicMock, payment_check: mock.Mock, caplog: pytest.LogCaptureFixture +): + """Recursively isolate one bad address without separating its original UTxOs.""" + groups = _payment_inputs(["a", "bad", "c", "d"]) + original, _ = cleanup.flatten_tx_inputs(groups) + cluster.g_query.get_utxo.return_value = original + cluster.g_transaction.send_tx.side_effect = _payment_sender({"bad"}) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=groups, faucet_address="f", tx_name="isolate" + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert len(calls) == 5 + successful = [ + c.kwargs["txins"] for c in calls if all(u.address != "bad" for u in c.kwargs["txins"]) + ] + assert {u.address for txins in successful for u in txins} == {"a", "c", "d"} + for call in calls: + txins = call.kwargs["txins"] + for address in {u.address for u in txins}: + assert [u for u in txins if u.address == address] == [ + u for u in original if u.address == address + ] + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) + payment_check.assert_called_once_with( + cluster_obj=cluster, txins=[txins[0] for txins in successful] + ) + assert "addresses=['bad']" in caplog.text + assert "tx_bad#0" in caplog.text + assert "tx_bad#1" in caplog.text + assert "MissingVKeyWitnessesUTXOW" in caplog.text + cluster.wait_for_new_block.assert_not_called() + + +def test_payment_rejection_continues_subsequent_batches( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """A failing single-address batch does not prevent the next batch succeeding.""" + groups = _payment_inputs(["bad", "good"], per_address=100) + cluster.g_transaction.send_tx.side_effect = _payment_sender({"bad"}) + cleanup.return_funds_to_faucet(cluster_obj=cluster, tx_inputs=groups, faucet_address="f") + assert cluster.g_transaction.send_tx.call_count == 2 + cluster.g_query.get_utxo.assert_not_called() + payment_check.assert_called_once_with(cluster_obj=cluster, txins=[groups[1][0][0]]) + + +def test_payment_refresh_preserves_original_set(cluster: mock.MagicMock, payment_check: mock.Mock): + """Remove stale inputs without sweeping new outputs returned by a refresh query.""" + groups = _payment_inputs(["stale", "good", "bad"]) + original, _ = cleanup.flatten_tx_inputs(groups) + extra = dataclasses.replace(groups[1][0][0], utxo_hash="new_output") + still_unspent = [groups[1][0][0], *groups[2][0], extra] + cluster.g_query.get_utxo.return_value = still_unspent + cluster.g_transaction.send_tx.side_effect = _payment_sender({"bad"}) + cleanup.return_funds_to_faucet(cluster_obj=cluster, tx_inputs=groups, faucet_address="f") + for call in cluster.g_query.get_utxo.call_args_list: + assert "address" not in call.kwargs + assert set(call.kwargs["txin"]) <= {f"{u.utxo_hash}#{u.utxo_ix}" for u in original} + successful = [ + c.kwargs["txins"] + for c in cluster.g_transaction.send_tx.call_args_list + if all(u.address != "bad" for u in c.kwargs["txins"]) + ] + assert successful == [[groups[1][0][0]]] + assert all(extra not in c.kwargs["txins"] for c in cluster.g_transaction.send_tx.call_args_list) + payment_check.assert_called_once_with(cluster_obj=cluster, txins=[groups[1][0][0]]) + + +def test_payment_refresh_drops_empty_groups(cluster: mock.MagicMock, payment_check: mock.Mock): + """Do not submit empty children after all original inputs disappear.""" + groups = _payment_inputs(["a", "b"]) + cluster.g_transaction.send_tx.side_effect = _payment_sender({"a"}) + cluster.g_query.get_utxo.return_value = [] + cleanup.return_funds_to_faucet(cluster_obj=cluster, tx_inputs=groups, faucet_address="f") + cluster.g_transaction.send_tx.assert_called_once() + payment_check.assert_not_called() + + +@pytest.mark.parametrize( + "error", + [ + subprocess.TimeoutExpired("transaction submit", 300), + TimeoutError("submission timed out"), + clusterlib.CLIError("MempoolTxTooSlow retry budget exhausted"), + clusterlib.CLIError("BadInputsUTxO"), + clusterlib.CLIError("AllInputsAreSpent"), + clusterlib.CLIError("Transaction didn't make it to the chain"), + clusterlib.CLIError("Failed to resubmit the transaction"), + clusterlib.CLIError("MuxError resource vanished (Broken pipe)"), + clusterlib.CLIError("Network.Socket.connect: resource exhausted"), + clusterlib.CLIError("socket connection refused"), + OSError("socket closed"), + clusterlib.CLIError("ValueNotConservedUTxO"), + clusterlib.CLIError("unknown submission failure"), + clusterlib.CLIError("MempoolTxTooSlow; Last rejection: FeeTooSmallUTxO"), + ], +) +def test_payment_uncertain_or_systemic_no_split( + cluster: mock.MagicMock, payment_check: mock.Mock, error: Exception +): + """Never recursively retry infrastructure, ambiguous or systemic failures.""" + cluster.g_transaction.send_tx.side_effect = error + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=_payment_inputs(["a", "b", "c"]), faucet_address="f" + ) + cluster.g_transaction.send_tx.assert_called_once() + cluster.g_query.get_utxo.assert_not_called() + payment_check.assert_not_called() + + +def test_payment_chained_timeout_no_retry(cluster: mock.MagicMock, payment_check: mock.Mock): + """A wrapper mentioning a ledger failure cannot hide an uncertain timeout cause.""" + error = clusterlib.CLIError("FeeTooSmallUTxO") + error.__cause__ = subprocess.TimeoutExpired("transaction submit", 300) + cluster.g_transaction.send_tx.side_effect = error + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=_payment_inputs(["a", "b"]), faucet_address="f" + ) + cluster.g_transaction.send_tx.assert_called_once() + payment_check.assert_not_called() + + +def test_payment_unknown_wrapper_not_definite_rejection( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """Do not classify an unknown wrapper using an earlier rejection in its cause.""" + error = clusterlib.CLIError("unknown submission outcome") + error.__cause__ = clusterlib.CLIError("FeeTooSmallUTxO") + cluster.g_transaction.send_tx.side_effect = error + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=_payment_inputs(["a", "b"]), faucet_address="f" + ) + cluster.g_transaction.send_tx.assert_called_once() + payment_check.assert_not_called() + + +def test_payment_fee_retries_bounded(cluster: mock.MagicMock, payment_check: mock.Mock): + """Exhaust fee retries once, rather than multiplying systemic failures by address.""" + cluster.g_transaction.send_tx.side_effect = clusterlib.CLIError("FeeTooSmallUTxO") + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=_payment_inputs(["a", "b", "c"]), faucet_address="f" + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert [c.kwargs["witness_count_add"] for c in calls] == [2, 7, 12] + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) + cluster.g_query.get_utxo.assert_not_called() + payment_check.assert_not_called() + + +def test_payment_fee_retries_and_splits_have_unique_names( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """Preserve bounded fee adjustments independently for parent and child groups.""" + groups = _payment_inputs(["good", "bad"]) + original, _ = cleanup.flatten_tx_inputs(groups) + cluster.g_query.get_utxo.return_value = original + cluster.g_transaction.send_tx.side_effect = [ + clusterlib.CLIError("FeeTooSmallUTxO"), + clusterlib.CLIError("MissingVKeyWitnessesUTXOW"), + clusterlib.CLIError("FeeTooSmallUTxO"), + mock.Mock(txins=groups[0][0]), + clusterlib.CLIError("MissingVKeyWitnessesUTXOW"), + ] + cleanup.return_funds_to_faucet(cluster_obj=cluster, tx_inputs=groups, faucet_address="f") + calls = cluster.g_transaction.send_tx.call_args_list + assert [c.kwargs["witness_count_add"] for c in calls] == [2, 7, 2, 7, 2] + assert len({c.kwargs["tx_name"] for c in calls}) == len(calls) + payment_check.assert_called_once_with(cluster_obj=cluster, txins=[groups[0][0][0]]) + + +def test_payment_no_reuse_after_uncertain_submission( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """Quarantine original IDs even when another signing-key group repeats them.""" + groups = _payment_inputs(["a"], per_address=100) + repeated = (groups[0][0], pl.Path("alias.skey")) + good = _payment_inputs(["good"]) + cluster.g_transaction.send_tx.side_effect = [ + subprocess.TimeoutExpired("submit", 300), + mock.Mock(txins=good[0][0]), + ] + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[*groups, repeated, *good], faucet_address="f" + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert len(calls) == 2 + assert calls[1].kwargs["txins"] == good[0][0] + payment_check.assert_called_once_with(cluster_obj=cluster, txins=[good[0][0][0]]) + + +def test_payment_refresh_does_not_reintroduce_submitted_inputs( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """An overbroad refresh cannot bring inputs from an earlier submitted batch back.""" + first = _payment_inputs(["first"], per_address=100) + later = _payment_inputs(["good", "bad"]) + all_inputs, _ = cleanup.flatten_tx_inputs([*first, *later]) + cluster.g_query.get_utxo.return_value = all_inputs + cluster.g_transaction.send_tx.side_effect = _payment_sender({"bad"}) + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=[*first, *later], faucet_address="f" + ) + calls = cluster.g_transaction.send_tx.call_args_list + assert all(u.address != "first" for c in calls[1:] for u in c.kwargs["txins"]) + payment_check.assert_called_once_with( + cluster_obj=cluster, txins=[first[0][0][0], later[0][0][0]] + ) + + +def test_payment_refresh_infrastructure_failure_stops_branch( + cluster: mock.MagicMock, payment_check: mock.Mock +): + """A failed refresh stops isolation rather than triggering recursive retry storms.""" + cluster.g_transaction.send_tx.side_effect = _payment_sender({"bad"}) + cluster.g_query.get_utxo.side_effect = clusterlib.CLIError("connection refused") + cleanup.return_funds_to_faucet( + cluster_obj=cluster, tx_inputs=_payment_inputs(["good", "bad"]), faucet_address="f" + ) + cluster.g_transaction.send_tx.assert_called_once() + cluster.g_query.get_utxo.assert_called_once() + payment_check.assert_not_called() From 28fa42f89f414d3ec0e1ff341079f82155b618a3 Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 6 Nov 2026 19:19:01 +0100 Subject: [PATCH 4/4] fix: isolate missing script witness failures Allow MissingScriptWitnessesUTXOW to trigger payment batch isolation so script-locked inputs do not block unrelated address cleanup. Add a mocked regression proving the normal address is recovered while the script-controlled address is isolated and logged. --- cardano_node_tests/utils/testnet_cleanup.py | 1 + framework_tests/test_testnet_cleanup.py | 25 +++++++++++++++++++-- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/cardano_node_tests/utils/testnet_cleanup.py b/cardano_node_tests/utils/testnet_cleanup.py index cc64a0bf9..c3f5f77a0 100644 --- a/cardano_node_tests/utils/testnet_cleanup.py +++ b/cardano_node_tests/utils/testnet_cleanup.py @@ -318,6 +318,7 @@ def _payment_rejection_reason(error: clusterlib.CLIError) -> str: reasons = ( "FeeTooSmallUTxO", "MissingVKeyWitnessesUTXOW", + "MissingScriptWitnessesUTXOW", "InvalidWitnessesUTXOW", "MaxTxSizeUTxO", "OutputTooSmallUTxO", diff --git a/framework_tests/test_testnet_cleanup.py b/framework_tests/test_testnet_cleanup.py index b53b6f010..5c6cd1fb6 100644 --- a/framework_tests/test_testnet_cleanup.py +++ b/framework_tests/test_testnet_cleanup.py @@ -539,12 +539,14 @@ def _payment_inputs(names: list[str], *, per_address: int = 2) -> cleanup.TxInpu ] -def _payment_sender(bad: set[str]) -> tp.Callable[..., mock.Mock]: +def _payment_sender( + bad: set[str], *, reason: str = "MissingVKeyWitnessesUTXOW" +) -> tp.Callable[..., mock.Mock]: """Reject any transaction containing a specified address, without submitting it.""" def send(*, txins: list[clusterlib.UTXOData], **_kwargs: tp.Any) -> mock.Mock: if any(u.address in bad for u in txins): - msg = "ShelleyTxValidationError (MissingVKeyWitnessesUTXOW bad_key)" + msg = f"ShelleyTxValidationError ({reason} bad_key)" raise clusterlib.CLIError(msg) return mock.Mock(txins=txins) @@ -593,6 +595,25 @@ def test_payment_rejection_recovers_unaffected( cluster.wait_for_new_block.assert_not_called() +def test_missing_script_witness_recovers_normal_address( + cluster: mock.MagicMock, payment_check: mock.Mock, caplog: pytest.LogCaptureFixture +): + """Isolate script-locked inputs without abandoning a normal address's funds.""" + groups = _payment_inputs(["normal", "script_locked"]) + original, _ = cleanup.flatten_tx_inputs(groups) + cluster.g_query.get_utxo.return_value = original + cluster.g_transaction.send_tx.side_effect = _payment_sender( + {"script_locked"}, reason="MissingScriptWitnessesUTXOW" + ) + cleanup.return_funds_to_faucet(cluster_obj=cluster, tx_inputs=groups, faucet_address="f") + calls = cluster.g_transaction.send_tx.call_args_list + assert [c.kwargs["txins"] for c in calls] == [original, groups[0][0], groups[1][0]] + payment_check.assert_called_once_with(cluster_obj=cluster, txins=[groups[0][0][0]]) + assert "addresses=['script_locked']" in caplog.text + assert "MissingScriptWitnessesUTXOW" in caplog.text + cluster.wait_for_new_block.assert_not_called() + + def test_payment_rejection_continues_subsequent_batches( cluster: mock.MagicMock, payment_check: mock.Mock ):