From 900f58e6f89a4a97b139869b5f4c6344dbc7d434 Mon Sep 17 00:00:00 2001 From: Martin Kourim Date: Fri, 13 Nov 2026 18:30:52 +0100 Subject: [PATCH] test: cover Plutus spends in certified EBs Add witness-script and reference-script subtests sharing one locked cluster and funded setup. Correlate exact transaction hashes and sizes through EB forging, certification, and adoption before checking script effects and retained collateral/reference inputs. Preserve settlement and fork-inconclusive checks, and document the EB reference helper's support for known-valid Dijkstra script spends. This does not claim EB size or execution-budget enforcement. Both cases passed twice on node/CLI b004fe0c5. Lint passed. --- .../tests/tests_dijkstra/leios.py | 11 +- .../tests/tests_dijkstra/test_leios_plutus.py | 367 ++++++++++++++++++ 2 files changed, 374 insertions(+), 4 deletions(-) create mode 100644 cardano_node_tests/tests/tests_dijkstra/test_leios_plutus.py diff --git a/cardano_node_tests/tests/tests_dijkstra/leios.py b/cardano_node_tests/tests/tests_dijkstra/leios.py index 91eb49ddb..ba23b7511 100644 --- a/cardano_node_tests/tests/tests_dijkstra/leios.py +++ b/cardano_node_tests/tests/tests_dijkstra/leios.py @@ -423,17 +423,20 @@ def is_committee_seated_in_genesis(*, cluster_obj: clusterlib.ClusterLib) -> boo def get_eb_tx_ref(*, tx_file: pl.Path) -> tuple[str, int]: - """Return the EB reference to a valid Dijkstra payment transaction. + """Return the EB reference to a known-valid top-level Dijkstra transaction. An EB doesn't list its txs by tx id (the hash of the tx body), but by the Blake2b-256 hash and size of the block-format tx (`serializeTxForLeios` in ouroboros-consensus `LeiosDemoTypes.hs`). The signed file holds the three-element mempool format `[body, witnesses, auxiliary_data]`. Block inclusion adds a trailing `is_valid` flag, - which is `true` for the payment txs this helper supports. Only the outer array header - is replaced and the flag appended, preserving the original bytes of every component. + which is `true` for payments and successful Plutus spends. The same encoding applies + to script witnesses and reference-script spends (`toCBORForBlockInclusion` in the + Dijkstra ledger). Only the outer array header is replaced and the flag appended, + preserving the original bytes of every component. Do not use this helper for + phase-2-invalid transactions: the mempool bytes do not encode the producer's flag. Args: - tx_file: A path to a signed valid Dijkstra payment tx (a text envelope). + tx_file: A path to a signed, known-valid Dijkstra tx (a text envelope). Returns: The block-format tx hash as a hex string, and its size in bytes. diff --git a/cardano_node_tests/tests/tests_dijkstra/test_leios_plutus.py b/cardano_node_tests/tests/tests_dijkstra/test_leios_plutus.py new file mode 100644 index 000000000..11739a3e5 --- /dev/null +++ b/cardano_node_tests/tests/tests_dijkstra/test_leios_plutus.py @@ -0,0 +1,367 @@ +"""Trace valid Plutus spends through adopted certified Leios endorser blocks.""" + +import dataclasses +import logging +import pathlib as pl +import time + +import allure +import pytest +import pytest_subtests +from cardano_clusterlib import clusterlib + +from cardano_node_tests.cluster_management import cluster_management +from cardano_node_tests.tests import addrs_common +from cardano_node_tests.tests import common +from cardano_node_tests.tests import markers +from cardano_node_tests.tests import plutus_common +from cardano_node_tests.tests.tests_dijkstra import leios +from cardano_node_tests.tests.tests_dijkstra import test_leios_blocks as eb_tests +from cardano_node_tests.utils import cluster_nodes +from cardano_node_tests.utils import dbsync_utils +from cardano_node_tests.utils import helpers +from cardano_node_tests.utils import logfiles + +LOGGER = logging.getLogger(__name__) + +pytestmark = [ + markers.SKIPIF_ON_TESTNET, + pytest.mark.skipif(bool(leios.SKIP_REASON), reason=leios.SKIP_REASON), + markers.SKIPIF_PLUTUSV2_UNUSABLE, +] + +NUM_TXS = 30 +MAX_ROUNDS = 3 +ROUND_TIMEOUT_SEC = 2 * leios.MAX_SEARCH_SEC +SCRIPT_AMOUNT = 10_000_000 + + +@dataclasses.dataclass(frozen=True) +class _ScriptTx: + """A signed script spend, its ledger identity, and its EB wire reference size.""" + + signed_file: pl.Path + tx_id: str + wire_size: int + raw_output: clusterlib.TxRawOutput + script_input: clusterlib.UTXOData + + +def _check_script_effects( + *, + cluster_obj: clusterlib.ClusterLib, + eb: eb_tests._CertifiedEb, + records: dict[str, _ScriptTx], + destination: str, + amount: int, + collaterals: list[clusterlib.UTXOData], + reference: clusterlib.UTXOData | None, +) -> None: + """Check the selected EB's script effects and unchanged collateral/reference inputs.""" + txins = [f"{records[h].tx_id}#0" for h in sorted(eb.tx_hashes)] + outputs = cluster_obj.g_query.get_utxo(txin=txins) + assert {f"{u.utxo_hash}#{u.utxo_ix}" for u in outputs} == set(txins), outputs + assert all(u.address == destination and u.amount == amount for u in outputs), outputs + spent = [records[h].script_input for h in sorted(eb.tx_hashes)] + assert not cluster_obj.g_query.get_utxo(utxo=spent), "Certified script inputs remain unspent" + + # Every transaction in the batch is valid, not only those selected for the EB claim. + retained = cluster_obj.g_query.get_utxo(utxo=collaterals) + assert sorted(retained) == sorted(collaterals), "A successful spend consumed collateral" + if reference is not None: + assert cluster_obj.g_query.get_utxo(utxo=reference) == [reference], ( + "The reference-script UTxO changed or was consumed" + ) + + +def _check_no_forks( + *, + searches: dict[pl.Path, logfiles.LogCursor], + regex: str, + lines_per_log: dict[pl.Path, list[str]], + errors: dict[pl.Path, str], + eb: eb_tests._CertifiedEb, +) -> None: + """Retain the payment oracle's inconclusive outcome on post-adoption fork switches.""" + eb_tests._collect_lines( + searches=searches, regex=regex, log_errors=errors, lines_per_log=lines_per_log + ) + forks = eb_tests._get_fork_switches_after(lines_per_log=lines_per_log, eb=eb) + if errors or forks: + pytest.skip(f"Script transaction's certified-EB path is inconclusive: {errors}; {forks}") + + +class TestLeiosPlutus: + """Known script spends included in certified EBs, not merely ordinary ranking blocks.""" + + @allure.link(helpers.get_vcs_link()) + @pytest.mark.leios + @pytest.mark.plutus + @pytest.mark.long + @pytest.mark.dbsync + @pytest.mark.xdist_split(markers.XdSplits.heavy) + def test_plutus_spends_in_certified_eb( + self, + cluster_singleton: clusterlib.ClusterLib, + cluster_manager: cluster_management.ClusterManager, + subtests: pytest_subtests.SubTests, + ): + """Require witness-script and reference-script spends in adopted certified EBs. + + Share the funded addresses and whole-cluster lock between both cases. For + each, create independent script inputs with inline datums and collateral, + then burst-submit large, valid Plutus V2 spends under Leios load. Match the + signed transactions' exact block-format hashes/sizes to a forged EB, its + certificate, adoption by the certifying producer, and another pool's tip. + Retry with fresh inputs when a batch is included only in ordinary RBs. + + Check the certified script inputs were consumed, expected payment outputs + exist, and collateral/reference UTxOs are unchanged, before and after block + settlement. Retain the payment oracle's fork/log-read inconclusive outcome. + This tests positive inclusion, not EB size or execution-budget enforcement. + """ + cluster = cluster_singleton + template = common.get_test_id(cluster) + leios.skip_if_no_ebs(cluster_obj=cluster) + seated = leios.is_committee_seated_in_genesis(cluster_obj=cluster) + # Committee activation plus bounded block waits in funding and settlement. + common.skip_on_long_epochs( + cluster_obj=cluster, epochs=3 + (0 if seated else leios.VOTING_START_EPOCH) + ) + if not seated: + cluster.wait_for_epoch(epoch_no=leios.VOTING_START_EPOCH, padding_seconds=5) + + addrs = addrs_common.get_payment_addrs( + name_template=f"{template}_payments", + cluster_manager=cluster_manager, + cluster_obj=cluster, + num=2, + fund_idx=[0], + amount=4_000_000_000, + caching_key=helpers.get_current_line_str(), + ) + metadata_file = eb_tests._get_big_metadata_file(name_template=template) + for use_reference in (False, True): + case = "reference_script" if use_reference else "witness_script" + with subtests.test(script_source=case): + self._check_script_case( + cluster_obj=cluster, + payment=addrs[0], + destination=addrs[1].address, + name_template=f"{template}_{case}", + metadata_file=metadata_file, + use_reference=use_reference, + ) + + def _check_script_case( + self, + *, + cluster_obj: clusterlib.ClusterLib, + payment: clusterlib.AddressRecord, + destination: str, + name_template: str, + metadata_file: pl.Path, + use_reference: bool, + ) -> None: + """Burst independent spends and prove a specific adopted certified-EB inclusion.""" + cluster = cluster_obj + script = plutus_common.ALWAYS_SUCCEEDS["v2"] + cost = script.execution_cost + script_address = cluster.g_address.gen_payment_addr( + addr_name=name_template, payment_script_file=script.script_file + ) + pparams = cluster.g_query.get_protocol_params() + # Cover max-size bytes, the small script budget, and reference-script fees. + fee = pparams["txFeePerByte"] * pparams["maxTxSize"] + pparams["txFeeFixed"] + 1_000_000 + assert fee < SCRIPT_AMOUNT, pparams + collateral_amount = max(5_000_000, (fee * pparams["collateralPercentage"] + 99) // 100) + state_dir = cluster_nodes.get_cluster_env().state_dir + pool_logs = sorted(state_dir.glob("pool*.stdout")) + assert pool_logs, state_dir + node_logs = [*pool_logs, state_dir / "bft1.stdout"] + regex = "|".join( + f"(?:{r})" + for r in ( + leios.EB_FORGED_RE, + leios.EB_ANNOUNCED_RE, + leios.EB_CERTIFIED_RE, + leios.BLOCK_ADOPTED_RE, + leios.NEW_TIP_RE, + ) + ) + + for round_no in range(MAX_ROUNDS): + template = f"{name_template}_r{round_no}" + # Keep each script input and collateral distinct, even at the same address. + txouts = [ + clusterlib.TxOut( + address=script_address, + amount=SCRIPT_AMOUNT, + inline_datum_file=plutus_common.DATUM_42, + ) + ] * NUM_TXS + [ + clusterlib.TxOut(address=payment.address, amount=collateral_amount) + ] * NUM_TXS + if use_reference: + txouts.append( + clusterlib.TxOut( + address=payment.address, + amount=10_000_000, + reference_script_file=script.script_file, + ) + ) + funding = cluster.g_transaction.send_tx( + src_address=payment.address, + tx_name=f"{template}_fund_scripts", + txouts=txouts, + tx_files=clusterlib.TxFiles(signing_key_files=[payment.skey_file]), + join_txouts=False, + ) + funded = {u.utxo_ix: u for u in cluster.g_query.get_utxo(tx_raw_output=funding)} + script_inputs = [funded[i] for i in range(NUM_TXS)] + collaterals = [funded[NUM_TXS + i] for i in range(NUM_TXS)] + reference = funded[2 * NUM_TXS] if use_reference else None + assert all( + u.address == script_address and u.inline_datum is not None for u in script_inputs + ) + assert all(u.amount == collateral_amount for u in collaterals), collaterals + if reference is not None: + assert reference.reference_script, reference + dbsync_utils.check_tx(cluster_obj=cluster, tx_raw_output=funding) + + # Build every spend before submitting, so metadata-backed load reaches an EB. + records: dict[str, _ScriptTx] = {} + for i, utxo in enumerate(script_inputs): + tx_name = f"{template}_spend{i}" + raw = cluster.g_transaction.build_raw_tx_bare( + out_file=f"{tx_name}.body", + txouts=[clusterlib.TxOut(address=destination, amount=SCRIPT_AMOUNT - fee)], + tx_files=clusterlib.TxFiles( + metadata_json_files=[metadata_file], + signing_key_files=[payment.skey_file], + ), + script_txins=[ + clusterlib.ScriptTxIn( + txins=[utxo], + script_file="" if use_reference else script.script_file, + reference_txin=reference, + reference_type=clusterlib.ScriptTypes.PLUTUS_V2 + if use_reference + else "", + inline_datum_present=True, + redeemer_cbor_file=plutus_common.REDEEMER_42_CBOR, + execution_units=(cost.per_time, cost.per_space), + collaterals=[collaterals[i]], + ) + ], + fee=fee, + ) + signed = cluster.g_transaction.sign_tx( + tx_body_file=raw.out_file, + signing_key_files=[payment.skey_file], + tx_name=tx_name, + ) + tx_hash, size = leios.get_eb_tx_ref(tx_file=signed) + records[tx_hash] = _ScriptTx( + signed_file=signed, + tx_id=cluster.g_transaction.get_txid(tx_file=signed), + wire_size=size, + raw_output=raw, + script_input=utxo, + ) + + searches = logfiles.init_searches(node_logs) + lines: dict[pl.Path, list[str]] = {p: [] for p in node_logs} + errors: dict[pl.Path, str] = {} + refs = {h: rec.wire_size for h, rec in records.items()} + for rec in records.values(): + cluster.g_transaction.submit_tx_bare(tx_file=rec.signed_file) + + # Ordinary RB inclusion cannot satisfy the correlated EB lifecycle oracle. + deadline = time.monotonic() + ROUND_TIMEOUT_SEC + certified: list[eb_tests._CertifiedEb] = [] + round_outputs = [f"{rec.tx_id}#0" for rec in records.values()] + while True: + eb_tests._collect_lines( + searches=searches, regex=regex, log_errors=errors, lines_per_log=lines + ) + certified, problems = eb_tests._find_certified_ebs( + lines_per_pool={p: lines[p] for p in pool_logs}, tx_refs=refs + ) + assert not problems, problems + if len(cluster.g_query.get_utxo(txin=round_outputs)) == NUM_TXS: + # Let final certification/adoption traces catch up before retrying. + cluster.wait_for_new_block(new_blocks=1) + eb_tests._collect_lines( + searches=searches, regex=regex, log_errors=errors, lines_per_log=lines + ) + certified, problems = eb_tests._find_certified_ebs( + lines_per_pool={p: lines[p] for p in pool_logs}, tx_refs=refs + ) + assert not problems, problems + break + assert time.monotonic() < deadline, ( + f"Script spends did not reach the ledger within {ROUND_TIMEOUT_SEC}s: {errors}" + ) + time.sleep(leios.SEARCH_STEP_SEC) + if not certified: + assert not errors, errors + LOGGER.info("No adopted certified script EB in %s; retrying", template) + continue + + eb = max(certified, key=lambda rec: len(rec.tx_hashes)) + _check_no_forks( + searches=searches, regex=regex, lines_per_log=lines, errors=errors, eb=eb + ) + helpers.write_json( + out_file=pl.Path(f"{template}_certified_eb.json"), + content={ + "use_reference_script": use_reference, + "eb_slot": eb.eb_slot, + "eb_hash": eb.eb_hash, + "cert_slot": eb.cert_slot, + "cert_rb_hash": eb.cert_rb_hash, + "destination": destination, + "amount": SCRIPT_AMOUNT - fee, + "transactions": [ + {"tx_id": records[h].tx_id, "eb_tx_hash": h, "wire_size": refs[h]} + for h in sorted(eb.tx_hashes) + ], + }, + ) + _check_script_effects( + cluster_obj=cluster, + eb=eb, + records=records, + destination=destination, + amount=SCRIPT_AMOUNT - fee, + collaterals=collaterals, + reference=reference, + ) + cluster.wait_for_new_block( + new_blocks=min(cluster.genesis["securityParam"] + 1, eb_tests.MAX_SETTLE_BLOCKS) + ) + _check_no_forks( + searches=searches, regex=regex, lines_per_log=lines, errors=errors, eb=eb + ) + _check_script_effects( + cluster_obj=cluster, + eb=eb, + records=records, + destination=destination, + amount=SCRIPT_AMOUNT - fee, + collaterals=collaterals, + reference=reference, + ) + for tx_hash in sorted(eb.tx_hashes): + dbsync_utils.check_tx( + cluster_obj=cluster, tx_raw_output=records[tx_hash].raw_output + ) + LOGGER.info( + "Certified EB %s contains %s script spends", + (eb.eb_slot, eb.eb_hash), + len(eb.tx_hashes), + ) + return + + pytest.fail(f"No script spend in an adopted certified EB after {MAX_ROUNDS} batches")