diff --git a/tests/cleanup_iptables.py b/tests/cleanup_iptables.py index a4b670e8224..fb56787d4f1 100644 --- a/tests/cleanup_iptables.py +++ b/tests/cleanup_iptables.py @@ -13,7 +13,7 @@ ) if len(sys.argv) > 1 and sys.argv[1] in ["-d", "--dump"]: - infra.partitions.Partitioner.dump() + infra.partitions.Partitioner.dump_all() else: - infra.partitions.Partitioner.dump() - infra.partitions.Partitioner.cleanup() + infra.partitions.Partitioner.dump_all() + infra.partitions.Partitioner.cleanup_all() diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index e5257f3af49..dec9d8f1653 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -3766,9 +3766,9 @@ def test_join_idempotency_short_circuits_on_backup(network, args): network.consortium.retire_node_by_id(primary, joined_node_id) -def run_backup_snapshot_download(const_args): +def _run_backup_snapshot_download(const_args, label_suffix, tests): args = copy.deepcopy(const_args) - args.label += "_backup_snapshot_download" + args.label += label_suffix # Use a small snapshot interval to trigger snapshots quickly args.snapshot_tx_interval = 30 args.nodes = infra.e2e_args.max_nodes(args, f=0) @@ -3780,11 +3780,41 @@ def run_backup_snapshot_download(const_args): txs=app.LoggingTxs("user0"), ) as network: network.start_and_open(args, backup_snapshot_fetch_enabled=True) - test_backup_snapshot_fetch(network, args) - test_backup_snapshot_fetch_max_size(network, args) - test_join_idempotency_short_circuits_on_backup(network, args) - test_join_time_snapshot_fetch_failure(network, args) - test_error_message_on_failure_to_fetch_snapshot(network, args) + for test in tests: + test(network, args) + + +# Each group below brings up its own network and runs concurrently with the +# others. Every test starts by finding the primary and issuing its own +# transactions, so none of them depend on the others. +def run_backup_snapshot_download(const_args): + _run_backup_snapshot_download( + const_args, + "_backup_snapshot_download", + [test_backup_snapshot_fetch], + ) + + +def run_backup_snapshot_download_limits(const_args): + _run_backup_snapshot_download( + const_args, + "_backup_snapshot_limits", + [ + test_backup_snapshot_fetch_max_size, + test_join_idempotency_short_circuits_on_backup, + ], + ) + + +def run_backup_snapshot_download_failures(const_args): + _run_backup_snapshot_download( + const_args, + "_backup_snapshot_failures", + [ + test_join_time_snapshot_fetch_failure, + test_error_message_on_failure_to_fetch_snapshot, + ], + ) def run_propose_request_vote(const_args): diff --git a/tests/governance.py b/tests/governance.py index 9b80999b88c..e458ecf2cb9 100644 --- a/tests/governance.py +++ b/tests/governance.py @@ -569,16 +569,26 @@ def gov(args): test_consensus_status(network, args) test_member_data(network, args) test_ack_state_digest_update(network, args) - network = test_all_members(network, args) - test_user(network, args) - test_jinja_templates(network, args) - test_no_quote(network, args) - test_node_data(network, args) - test_each_node_cert_renewal(network, args) - test_binding_proposal_to_service_identity(network, args) - test_all_nodes_cert_renewal(network, args) - test_service_cert_renewal(network, args) - test_service_cert_renewal_extended(network, args) + + # test_all_members stops this network and recovers into a new one, which + # the enclosing context manager does not own: it still holds the + # original. Stop the recovered network here, or its nodes outlive the + # test. That includes the deliberately untrusted nodes added by + # test_no_quote and test_node_data, which then sit in a join retry loop + # for the rest of the CI job. + recovered_network = test_all_members(network, args) + try: + test_user(recovered_network, args) + test_jinja_templates(recovered_network, args) + test_no_quote(recovered_network, args) + test_node_data(recovered_network, args) + test_each_node_cert_renewal(recovered_network, args) + test_binding_proposal_to_service_identity(recovered_network, args) + test_all_nodes_cert_renewal(recovered_network, args) + test_service_cert_renewal(recovered_network, args) + test_service_cert_renewal_extended(recovered_network, args) + finally: + recovered_network.stop_all_nodes(skip_verification=True) # These tests requiring starting up + shutting down a node with specific diff --git a/tests/infra/partitions.py b/tests/infra/partitions.py index 4272ba89239..8e175f0ea4e 100644 --- a/tests/infra/partitions.py +++ b/tests/infra/partitions.py @@ -1,7 +1,10 @@ # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the Apache 2.0 License. import enum +import itertools import json +import os +import threading from dataclasses import field import iptc @@ -10,12 +13,82 @@ import infra.network import infra.node -CCF_IPTABLES_CHAIN = "CCF-TEST" +# Each Partitioner owns its own chain, so that several partitioned networks can +# run concurrently in one container without flushing each other's rules. The +# prefix is shared so that leftovers from a killed run can all be found. +CCF_IPTABLES_CHAIN_PREFIX = "CCF-TEST" + +# iptables chain names are limited to 28 characters. +MAX_CHAIN_NAME_LENGTH = 28 + +_chain_counter = itertools.count() +_chain_counter_lock = threading.Lock() + +# libiptc reads the whole table, modifies it and writes it back, and +# iptc.easy operates on a table object that is shared process-wide and +# committed and refreshed on every call. Interleaving calls from several +# threads therefore loses updates: one thread's refresh can discard another's +# pending change, leaving stale DROP rules behind. Every access to the filter +# table goes through this lock, and callers hold it across a whole set of +# related rules so that a partition appears and disappears atomically. It is +# re-entrant so the helpers can be nested inside those wider sections. +_iptables_lock = threading.RLock() + + +def _next_chain_name(): + with _chain_counter_lock: + index = next(_chain_counter) + # The pid keeps chains distinct across concurrently running ctest processes, + # the counter across Partitioners within one process. + name = f"{CCF_IPTABLES_CHAIN_PREFIX}-{os.getpid()}-{index}" + if len(name) > MAX_CHAIN_NAME_LENGTH: + raise ValueError( + f"iptables chain name {name!r} is {len(name)} characters, " + f"but iptables allows at most {MAX_CHAIN_NAME_LENGTH}" + ) + return name + + +def _input_rule(chain_name): + return {"protocol": "tcp", "target": chain_name} + + +def _delete_chain(chain_name): + with _iptables_lock: + if iptc.easy.has_chain("filter", chain_name): + iptc.easy.flush_chain("filter", chain_name) + if iptc.easy.has_rule("filter", "INPUT", _input_rule(chain_name)): + iptc.easy.delete_rule("filter", "INPUT", _input_rule(chain_name)) + iptc.easy.delete_chain("filter", chain_name) + + +def _create_chain(chain_name): + with _iptables_lock: + iptc.easy.add_chain("filter", chain_name) + iptc.easy.insert_rule("filter", "INPUT", _input_rule(chain_name)) + + +def _replace_rule(chain_name, rule): + with _iptables_lock: + if iptc.easy.has_rule("filter", chain_name, rule): + iptc.easy.delete_rule("filter", chain_name, rule) + iptc.easy.insert_rule("filter", chain_name, rule) + + +def _drop_rule(chain_name, rule): + with _iptables_lock: + if iptc.easy.has_rule("filter", chain_name, rule): + iptc.easy.delete_rule("filter", chain_name, rule) + + +def _ccf_chains(): + with _iptables_lock: + return [ + chain + for chain in iptc.easy.get_chains("filter") + if chain.startswith(CCF_IPTABLES_CHAIN_PREFIX) + ] -CCF_INPUT_RULE = { - "protocol": "tcp", - "target": CCF_IPTABLES_CHAIN, -} # Note: When playing with iptables rules on a remote VM, you may want to: # 1. Save the current iptable rules: $ sudo iptables-save > /etc/iptables.conf @@ -43,9 +116,10 @@ class Rules: name: str | None = None - def __init__(self, rules, name=None): + def __init__(self, rules, name=None, chain_name=None): self.rules = rules self.name = name + self.chain_name = chain_name def __enter__(self): return self @@ -55,9 +129,13 @@ def __exit__(self, type_, value, traceback): def drop(self): LOG.info(f'Dropping rules "{self.name or "[unamed]"}"') - for rule in self.rules: - if iptc.easy.has_rule("filter", CCF_IPTABLES_CHAIN, rule): - iptc.easy.delete_rule("filter", CCF_IPTABLES_CHAIN, rule) + if self.chain_name is None: + return + # Drop the whole set in one locked section, so that a partition is never + # observed half-removed. + with _iptables_lock: + for rule in self.rules: + _drop_rule(self.chain_name, rule) class Partitioner: @@ -72,29 +150,55 @@ class Partitioner: Note: It should be managed by a :py:class:`infra.network.Network` instance so that rules outlive nodes to avoid spurious log messages when the network is shutdown. + + Each instance owns a private iptables chain, so several partitioned networks + may exist at once. Rules only ever match their own network's node addresses + and ports, so co-existing chains do not affect each other. """ - @staticmethod - def dump(): - if iptc.easy.has_chain("filter", CCF_IPTABLES_CHAIN): + def dump(self): + if iptc.easy.has_chain("filter", self.chain_name): chain_status = ( "active" - if iptc.easy.has_rule("filter", "INPUT", CCF_INPUT_RULE) + if iptc.easy.has_rule("filter", "INPUT", _input_rule(self.chain_name)) else "inactive" ) LOG.info( - f'Dumping {chain_status} chain {CCF_IPTABLES_CHAIN}:\n{json.dumps(iptc.easy.dump_chain("filter", CCF_IPTABLES_CHAIN), indent=2)}' + f'Dumping {chain_status} chain {self.chain_name}:\n{json.dumps(iptc.easy.dump_chain("filter", self.chain_name), indent=2)}' ) else: - LOG.info(f"Chain {CCF_IPTABLES_CHAIN} does not exist") + LOG.info(f"Chain {self.chain_name} does not exist") @staticmethod - def cleanup(): - if iptc.easy.has_chain("filter", CCF_IPTABLES_CHAIN): - iptc.easy.flush_chain("filter", CCF_IPTABLES_CHAIN) - iptc.easy.delete_rule("filter", "INPUT", CCF_INPUT_RULE) - iptc.easy.delete_chain("filter", CCF_IPTABLES_CHAIN) - LOG.info(f"{CCF_IPTABLES_CHAIN} iptables chain cleaned up") + def dump_all(): + chains = _ccf_chains() + if not chains: + LOG.info(f"No {CCF_IPTABLES_CHAIN_PREFIX} iptables chain exists") + return + for chain_name in chains: + chain_status = ( + "active" + if iptc.easy.has_rule("filter", "INPUT", _input_rule(chain_name)) + else "inactive" + ) + LOG.info( + f'Dumping {chain_status} chain {chain_name}:\n{json.dumps(iptc.easy.dump_chain("filter", chain_name), indent=2)}' + ) + + def cleanup(self): + _delete_chain(self.chain_name) + LOG.info(f"{self.chain_name} iptables chain cleaned up") + + @staticmethod + def cleanup_all(): + """Remove every chain this infrastructure may have left behind. + + Only safe to call when no partitioned network is running, so it is used + by tests/cleanup_iptables.py rather than by the test infrastructure. + """ + for chain_name in _ccf_chains(): + _delete_chain(chain_name) + LOG.info(f"{CCF_IPTABLES_CHAIN_PREFIX} iptables chains cleaned up") @staticmethod def reverse_rule(rule): @@ -119,15 +223,14 @@ def swap_fields(obj, a, b): def __init__(self, network): self.network = network + self.chain_name = _next_chain_name() - # Cleanup any leftover rules - self.cleanup() - - # Create iptables chain - iptc.easy.add_chain("filter", CCF_IPTABLES_CHAIN) + # Cleanup any leftover rules from a previous run that happened to reuse + # this name + _delete_chain(self.chain_name) - # Create iptables rule in INPUT chain - iptc.easy.insert_rule("filter", "INPUT", CCF_INPUT_RULE) + # Create iptables chain, and the INPUT rule that jumps into it + _create_chain(self.chain_name) def isolate_node( self, @@ -180,15 +283,15 @@ def isolate_node( if isolation_dir & IsolationDir.OUTBOUND_RESPONSES: rules.append(self.reverse_rule(client_rule)) - for rule in rules: - if iptc.easy.has_rule("filter", CCF_IPTABLES_CHAIN, rule): - iptc.easy.delete_rule("filter", CCF_IPTABLES_CHAIN, rule) - - iptc.easy.insert_rule("filter", CCF_IPTABLES_CHAIN, rule) + # Apply the whole set in one locked section, so that a partition is + # never observed half-applied. + with _iptables_lock: + for rule in rules: + _replace_rule(self.chain_name, rule) LOG.debug(name) - return Rules(rules, name) + return Rules(rules, name, self.chain_name) @staticmethod def _get_partition_name(partition: list[infra.node.Node]): @@ -231,19 +334,22 @@ def partition( rules = [] partitions_name = [] - for i, partition in enumerate(args): - partitions_name.append(f"{self._get_partition_name(partition)}") - # Rules are bi-directional so skip partitions that have already been enforced - other_partitions = args[i + 1 :] - - for node in partition: - for other_partition in other_partitions: - for other_node in other_partition: + # A partition is several isolate_node calls; hold the lock across all of + # them so the partition takes effect in one step. + with _iptables_lock: + for i, partition in enumerate(args): + partitions_name.append(f"{self._get_partition_name(partition)}") + # Rules are bi-directional so skip partitions that have already been enforced + other_partitions = args[i + 1 :] + + for node in partition: + for other_partition in other_partitions: + for other_node in other_partition: + rules.extend(self.isolate_node(node, other_node).rules) + + for other_node in other_nodes: rules.extend(self.isolate_node(node, other_node).rules) - for other_node in other_nodes: - rules.extend(self.isolate_node(node, other_node).rules) - partitions_name.append(self._get_partition_name(other_nodes)) # Override partition name if it is specified by the caller @@ -251,10 +357,10 @@ def partition( LOG.success(f"Created new partition {partition_name}") - return Rules(rules, partition_name) + return Rules(rules, partition_name, self.chain_name) def partitions(self, *args: list[list[infra.node.Node]]): - rule = Rules([]) + rule = Rules([], chain_name=self.chain_name) names = [] for nodes in args: r = self.partition(*nodes) diff --git a/tests/infra/runner.py b/tests/infra/runner.py index 105ed1f0485..be5b53c5e84 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -8,6 +8,7 @@ import sys import threading import time +from concurrent.futures import ThreadPoolExecutor, as_completed from random import seed from typing import ClassVar @@ -197,8 +198,6 @@ def log_exception(args: threading.ExceptHookArgs): class ConcurrentRunner: - threads: ClassVar[list[threading.Thread]] = [] - # Env var to filter sub-tests by exact name match. Value is a # '|'-separated list, e.g. CR_FILTER="testname1|testname2". When set, # only sub-tests whose name fully matches one of the entries are added. @@ -224,6 +223,10 @@ def add(parser): add_options(parser) self.args = infra.e2e_args.cli_args(add=add) + # Sub-tests to run, as (name, target, args) triples. Per instance, so + # that two runners in one process do not inherit each other's sub-tests. + # Threads are created by run(), so the pool decides how many exist. + self.tests: list[tuple[str, object, object]] = [] def add(self, prefix, target, **args_overrides): if self._test_filter is not None and prefix not in self._test_filter: @@ -232,7 +235,48 @@ def add(self, prefix, target, **args_overrides): for k, v in args_overrides.items(): setattr(args_, k, v) args_.label = f"{prefix}_{self.args.label}" - self.threads.append(threading.Thread(name=prefix, target=target, args=[args_])) + self.tests.append((prefix, target, args_)) + + @staticmethod + def default_max_concurrent(): + """Concurrent sub-tests a runner may have in flight. + + Each sub-test drives its own CCF network of one to five node processes. + Nodes spend most of their time waiting on timers and sockets, but a + network that cannot get CPU promptly sees spurious leadership elections + and dropped sessions, so this bounds how many run at once. + + One per core: a ceiling on new growth rather than a tightening of what + already worked, since the largest runner sustains around fifteen + concurrent nodes on a sixteen-core CI runner without trouble. Tests + whose sub-tests are unusually sensitive, such as partitions_test, pass + a lower value to run(). + """ + cores_count = len(os.sched_getaffinity(0)) + return max(2, cores_count) + + def _resolve_max_concurrent(self, max_concurrent): + limits = [max_concurrent or self.default_max_concurrent()] + + # Instrumented builds process every operation far more slowly, so they + # sustain fewer concurrent networks before nodes start missing their + # election timeouts. + if os.getenv("TSAN_OPTIONS") or os.getenv("CCF_GLIBCXX_DEBUG"): + cores_count = len(os.sched_getaffinity(0)) + avg_nodes_per_network = 3 + safety_factor = 0.5 + limits.append( + max(1, int(safety_factor * cores_count / avg_nodes_per_network)) + ) + + return max(1, min(limits)) + + @staticmethod + def _run_one(name, target, args): + # Sub-tests are identified by thread name in the log format, so restore + # it here: pool workers are reused and carry the previous name. + threading.current_thread().name = name + target(args) def run(self, max_concurrent=None): config = { @@ -247,50 +291,48 @@ def run(self, max_concurrent=None): } LOG.configure(**config) + tests = self.tests if self.args.regex: - self.threads = [ - thread - for thread in self.threads - if re.compile(self.args.regex).search(thread.name) - ] + pattern = re.compile(self.args.regex) + tests = [test for test in tests if pattern.search(test[0])] if self.args.show_only: - for thread in self.threads: - print(thread.name) + for name, _, _ in tests: + print(name) return - if not max_concurrent: - max_concurrent = len(self.threads) - - if os.getenv("TSAN_OPTIONS"): - cores_count = len(os.sched_getaffinity(0)) - avg_nodes_per_network = 3 - safety_factor = 0.5 - max_concurrent = int(safety_factor * cores_count / avg_nodes_per_network) - assert max_concurrent > 0 - - if os.getenv("CCF_GLIBCXX_DEBUG"): - # _GLIBCXX_DEBUG checks make every container op significantly - # slower, so a Debug build cannot sustain as many concurrent - # networks. Cap concurrency to avoid CPU starvation that - # manifests as spurious leadership elections / session loss. - cores_count = len(os.sched_getaffinity(0)) - avg_nodes_per_network = 3 - safety_factor = 0.5 - debug_cap = max(1, int(safety_factor * cores_count / avg_nodes_per_network)) - max_concurrent = min(max_concurrent, debug_cap) - - thread_groups = [ - self.threads[i : i + max_concurrent] - for i in range(0, len(self.threads), max_concurrent) - ] + if not tests: + return - for group in thread_groups: - for thread in group: - thread.start() + max_concurrent = self._resolve_max_concurrent(max_concurrent) + LOG.info( + f"Running {len(tests)} sub-tests, at most {max_concurrent} concurrently" + ) - for thread in group: - thread.join() + # A bounded pool rather than fixed batches: a sub-test starts as soon as + # any other finishes, so a single long sub-test does not hold back the + # ones queued behind it. + failures = [] + with ThreadPoolExecutor(max_workers=max_concurrent) as pool: + futures = { + pool.submit(self._run_one, name, target, args): name + for name, target, args in tests + } + for future in as_completed(futures): + name = futures[future] + try: + future.result() + except Exception as e: + description = f"Failure in {name}: {e!r}" + failures.append(description) + LOG.error( + description + + "\n" + + "".join(better_exceptions.format_exception(*sys.exc_info())) + ) - if FAILURES: - raise RuntimeError(FAILURES) + # FAILURES catches exceptions from threads the sub-tests start + # themselves, which do not surface through the pool's futures. + failures.extend(FAILURES) + if failures: + raise RuntimeError(failures) diff --git a/tests/nodes.py b/tests/nodes.py index 966fd17c048..7b9429d2ba0 100644 --- a/tests/nodes.py +++ b/tests/nodes.py @@ -322,4 +322,25 @@ def add(parser): nodes=infra.e2e_args.min_nodes(cr.args, f=1), ) + # Each of these builds its own single-node network, so they run + # concurrently with everything else. + for name, target in ( + ("join_old_snapshot", reconfiguration.run_join_old_snapshot), + ( + "join_no_snapshot", + reconfiguration.run_join_no_snapshot_against_original_primary, + ), + ("join_old_snapshot_ipv6", reconfiguration.run_join_old_snapshot_ipv6), + ( + "join_no_snapshot_ipv6", + reconfiguration.run_join_no_snapshot_against_original_primary_ipv6, + ), + ): + cr.add( + name, + target, + package="samples/apps/logging/logging", + nodes=infra.e2e_args.nodes(cr.args, 1), + ) + cr.run() diff --git a/tests/partitions_test.py b/tests/partitions_test.py index 8e4c743cc0f..e5a0eed44ee 100644 --- a/tests/partitions_test.py +++ b/tests/partitions_test.py @@ -24,6 +24,7 @@ from e2e_logging import verify_receipt from infra.checker import check_can_progress, check_does_not_progress from infra.log_capture import flush_info +from infra.runner import ConcurrentRunner from infra.tx_status import TxStatus from loguru import logger as LOG from reconfiguration import test_ledger_invariants @@ -1549,46 +1550,102 @@ def overhead(num_transactions, num_signatures): assert len(chunk_ends_to_expected_size) == 0 -def run(args): - txs = app.LoggingTxs("user0") +@contextlib.contextmanager +def partitioned_network(args): + """A fresh partitioned network for one group of tests. + Each group runs on its own network so that groups can run concurrently. + Every Partitioner owns a private iptables chain whose rules only match its + own nodes' addresses and ports, so co-existing groups do not interfere. + """ with infra.network.network( args.nodes, args.binary_dir, args.debug_nodes, pdb=args.pdb, - txs=txs, + txs=app.LoggingTxs("user0"), init_partitioner=True, ) as network: network.start_and_open(args) + yield network + +def run_basic_partitions(args): + with partitioned_network(args) as network: test_invalid_partitions(network, args) test_partition_majority(network, args) test_isolate_primary_from_one_backup(network, args) test_new_joiner_helps_liveness(network, args) + + +def run_certificate_partitions(args): + with partitioned_network(args) as network: test_expired_certs(network, args) test_rolled_back_node_certificate(network, args) + + +def run_isolate_and_reconnect(args): + with partitioned_network(args) as network: for n in range(5): test_isolate_and_reconnect_primary(network, args, iteration=n) + + +def run_reconfiguration_partitions(args): + with partitioned_network(args) as network: test_join_rollback_on_primary_isolation(network, args) test_election_reconfiguration(network, args) + + +def run_forwarding_and_sessions(args): + with partitioned_network(args) as network: test_forwarding_timeout(network, args) test_invalidated_blocking_calls(network, args) # HTTP2 doesn't support forwarding if not args.http2: test_session_consistency(network, args) - network = test_recovery_elections(network, args) - test_ledger_invariants(network, args) - run_ledger_chunk_bytes_check(args) - run_in_place_restart_uncommittable_ledger_check(args) + + +def run_recovery_elections(args): + with partitioned_network(args) as network: + # test_recovery_elections stops this network and recovers into a new + # one, which the context manager does not own: it still holds the + # original. Stop the returned network here, or its nodes outlive the + # test. + recovery_network = test_recovery_elections(network, args) + try: + test_ledger_invariants(recovery_network, args) + finally: + if recovery_network is not network: + recovery_network.stop_all_nodes(skip_verification=True) if __name__ == "__main__": - args = infra.e2e_args.cli_args() - args.nodes = infra.e2e_args.min_nodes(args, f=1) - args.package = "samples/apps/logging/logging" - args.snapshot_tx_interval = ( + cr = ConcurrentRunner() + cr.args.snapshot_tx_interval = ( 20 # Increase snapshot frequency for faster reconfigurations ) - run(args) + # Each group below runs on its own network, concurrently, and preserves the + # relative order of the tests it contains. + for name, target in ( + ("basic", run_basic_partitions), + ("certs", run_certificate_partitions), + ("isolate-reconnect", run_isolate_and_reconnect), + ("reconfiguration", run_reconfiguration_partitions), + ("forwarding", run_forwarding_and_sessions), + ("recovery-elections", run_recovery_elections), + ("ledger-chunks", run_ledger_chunk_bytes_check), + ("in-place-restart", run_in_place_restart_uncommittable_ledger_check), + ): + cr.add( + name, + target, + package="samples/apps/logging/logging", + nodes=infra.e2e_args.min_nodes(cr.args, f=1), + ) + + # These groups deliberately isolate nodes and wait for elections, so they + # are the most sensitive in the suite both to not getting CPU promptly and + # to contention on the shared iptables table: a starved or still-partitioned + # node looks like a failed election. Run few at once. + cr.run(max_concurrent=2) diff --git a/tests/reconfiguration.py b/tests/reconfiguration.py index 8293f6395e1..452464e5d41 100644 --- a/tests/reconfiguration.py +++ b/tests/reconfiguration.py @@ -1238,9 +1238,6 @@ def run_all(args, ipv6=False): if ipv6: assert_no_ipv4_in_node_configs(network) - run_join_old_snapshot(args, ipv6=ipv6) - run_join_no_snapshot_against_original_primary(args, ipv6=ipv6) - def run_join_old_snapshot(const_args, ipv6=False): txs = app.LoggingTxs("user0") @@ -1410,10 +1407,29 @@ def run_join_no_snapshot_against_original_primary(const_args, ipv6=False): ), f"Joiner should have started from a fetched snapshot, got startup_seqno={body['startup_seqno']}" -def run_ipv6(args): +def _assert_ipv6_available(): assert infra.net.ipv6_loopback_available(), ( "IPv6 loopback (::1) is not available; CI enables IPv6 via the " "container --sysctl net.ipv6.conf.*.disable_ipv6=0 (see .github/workflows)" ) + +def run_ipv6(args): + _assert_ipv6_available() + run_all(args, ipv6=True) + + +# Each of these builds its own single-node network and shares no state with +# run_all, so they are registered as their own sub-tests and run concurrently +# to minimise end-to-end test duration. +def run_join_old_snapshot_ipv6(args): + _assert_ipv6_available() + + run_join_old_snapshot(args, ipv6=True) + + +def run_join_no_snapshot_against_original_primary_ipv6(args): + _assert_ipv6_available() + + run_join_no_snapshot_against_original_primary(args, ipv6=True) diff --git a/tests/schema.py b/tests/schema.py index c32ee890a45..6793af01fe6 100644 --- a/tests/schema.py +++ b/tests/schema.py @@ -264,8 +264,7 @@ def add(parser): initial_member_count=1, ) - # The operations tests are split into groups which run concurrently, as the - # single sequential group used to dominate this test's total run time. + # These groups run concurrently, each on its own network. for name, target in ( ("operations-offline", e2e_operations.run_offline_ledger_tools), ("operations-snapshots", e2e_operations.run_snapshot_manual_and_retention), @@ -293,12 +292,23 @@ def add(parser): ledger_chunk_bytes="1B", # Chunk ledger at every signature transaction ) - cr.add( - "download-snapshot", - e2e_operations.run_backup_snapshot_download, - package="samples/apps/logging/logging", - nodes=infra.e2e_args.max_nodes(cr.args, f=0), - initial_user_count=1, - ) + for name, target in ( + ("download-snapshot", e2e_operations.run_backup_snapshot_download), + ( + "download-snapshot-limits", + e2e_operations.run_backup_snapshot_download_limits, + ), + ( + "download-snapshot-failures", + e2e_operations.run_backup_snapshot_download_failures, + ), + ): + cr.add( + name, + target, + package="samples/apps/logging/logging", + nodes=infra.e2e_args.max_nodes(cr.args, f=0), + initial_user_count=1, + ) cr.run()