diff --git a/cardano_node_tests/utils/testnet_cleanup.py b/cardano_node_tests/utils/testnet_cleanup.py index 17253a60d..c3f5f77a0 100644 --- a/cardano_node_tests/utils/testnet_cleanup.py +++ b/cardano_node_tests/utils/testnet_cleanup.py @@ -7,17 +7,22 @@ """ import concurrent.futures +import dataclasses import functools import itertools +import json import logging import pathlib as pl import random +import re +import subprocess import time import typing as tp 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 +31,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 +143,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 +176,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 +215,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 @@ -245,6 +284,160 @@ 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", + "MissingScriptWitnessesUTXOW", + "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, @@ -252,51 +445,36 @@ 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. - 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: - 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: - 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, + ) - 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 +535,208 @@ 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 _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 - def _run(files: list[pl.Path], f_addr: clusterlib.AddressRecord) -> TxInputGroup: - tx_inputs = [] - for fpath in files: - f_name = fpath.name - # Add random sleep for < 0.5s to prevent - # "Network.Socket.connect: : resource exhausted" - time.sleep(random.random() / 2) +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) + + 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", + ) - 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 - 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, - ) +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 + # 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: + 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") - deregister_stake_addr( + 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 + 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") - # 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] - - # 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 +745,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 +809,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 +914,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 +925,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..5c6cd1fb6 --- /dev/null +++ b/framework_tests/test_testnet_cleanup.py @@ -0,0 +1,803 @@ +"""Unit tests for batched testnet cleanup without a running node.""" + +import dataclasses +import json +import pathlib as pl +import subprocess +import typing as tp +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() + + +@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") + 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 [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() + + +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]) + 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): + """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 + + +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], *, 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 = f"ShelleyTxValidationError ({reason} 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_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 +): + """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()