From 5dfbcb43dd96418dffeee46b5d8e88509dea7d7b Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 16:07:32 +0100 Subject: [PATCH 1/6] Split the longest e2e tests into concurrent sub-tests The two virtual e2e jobs are the CI critical path, at 15.2 min (bucket_b) and 15.5 min (bucket_c plus partitions) against a 15 min target. Each pays ~5 min of fixed overhead, so the tests have to fit well under 10 min. ctest runs the tests in a bucket one after another, so a bucket costs the sum of its tests' wall times, and each test costs its longest sub-test. Measuring node process lifetimes from the logs of run 33500039436 shows the time is concentrated in a few long serial chains rather than spread out. `partitions` was the worst: 282s, and the only e2e test not using ConcurrentRunner at all. One network ran fourteen tests in sequence, then two further networks ran one after the other. Split it into eight sub-tests, each on its own network, keeping the original relative order within each group. `nodes_test` ran `run_join_old_snapshot` (22s) and `run_join_no_snapshot_against_original_primary` (7s) sequentially at the end of `reconfiguration.run_all`, once for IPv4 and once for IPv6. They each build their own single-node network and share no state with the main chain, so they become four sub-tests of their own. `schema_test` was paced by `download-snapshot` at 104s of its 105s total, a single network running five backup-snapshot-fetch tests in sequence. Each finds the primary and issues its own transactions, so split them into three groups. To run several partitioned networks at once, `Partitioner` now owns a private iptables chain named for the pid and an instance counter, rather than sharing a global `CCF-TEST` chain that every new instance flushed. Its rules already matched only its own nodes' addresses and ports, so co-existing chains do not interfere. libiptc reads, modifies and writes the whole table, so mutations take a process-wide lock. `cleanup_iptables.py` gains `cleanup_all` to remove leftovers from a killed run. ConcurrentRunner now runs sub-tests in a bounded pool rather than in fixed batches, and caps concurrency at half the cores by default. Batching made a cap expensive, because a single long sub-test held back everything queued behind it; a pool starts the next sub-test as soon as any finishes, so the cap costs nothing when a critical path dominates. This is what keeps the extra networks from overwhelming the runner. `partitions` asks for a lower cap, since tests that deliberately isolate nodes and wait for elections are the least tolerant of not getting CPU promptly. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 24ad6978-a80f-45bc-b73b-268aeaf2b30e --- tests/cleanup_iptables.py | 6 +- tests/e2e_operations.py | 45 +++++++++-- tests/infra/partitions.py | 161 +++++++++++++++++++++++++++++--------- tests/infra/runner.py | 123 +++++++++++++++++++---------- tests/nodes.py | 21 +++++ tests/partitions_test.py | 69 +++++++++++++--- tests/reconfiguration.py | 24 +++++- tests/schema.py | 25 ++++-- 8 files changed, 364 insertions(+), 110 deletions(-) diff --git a/tests/cleanup_iptables.py b/tests/cleanup_iptables.py index a4b670e82240..fb56787d4f1e 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 e5257f3af493..c5190f56ba86 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,42 @@ 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) + + +# These used to run as one sequential group, which made it the longest sub-test +# in schema_test by a wide margin. Each group below brings up its own network, +# so they run concurrently. 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/infra/partitions.py b/tests/infra/partitions.py index 4272ba892394..17a64fb0e384 100644 --- a/tests/infra/partitions.py +++ b/tests/infra/partitions.py @@ -1,21 +1,86 @@ # 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 infra.network +import infra.node import iptc from loguru import logger as LOG -import infra.network -import infra.node +# 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, so two threads +# committing at once can silently lose each other's rules. Every mutation of the +# filter table goes through this lock, which is enough because all partitioned +# networks in a test run live in one process (infra.runner.ConcurrentRunner +# threads). +_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}" + assert len(name) <= MAX_CHAIN_NAME_LENGTH, name + 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) + -CCF_IPTABLES_CHAIN = "CCF-TEST" +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 +108,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 +121,10 @@ def __exit__(self, type_, value, traceback): def drop(self): LOG.info(f'Dropping rules "{self.name or "[unamed]"}"') + if self.chain_name is None: + return for rule in self.rules: - if iptc.easy.has_rule("filter", CCF_IPTABLES_CHAIN, rule): - iptc.easy.delete_rule("filter", CCF_IPTABLES_CHAIN, rule) + _drop_rule(self.chain_name, rule) class Partitioner: @@ -72,29 +139,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 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(): - 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 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 +212,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() + # Cleanup any leftover rules from a previous run that happened to reuse + # this name + _delete_chain(self.chain_name) - # Create iptables chain - iptc.easy.add_chain("filter", CCF_IPTABLES_CHAIN) - - # 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, @@ -181,14 +273,11 @@ def isolate_node( 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) + _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]): @@ -251,10 +340,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 105ed1f04858..21ca38d206be 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -8,17 +8,17 @@ import sys import threading import time +from concurrent.futures import ThreadPoolExecutor, as_completed from random import seed from typing import ClassVar import better_exceptions -from loguru import logger as LOG - import infra.bencher import infra.jwt_issuer import infra.network import infra.proc import infra.remote_client +from loguru import logger as LOG logging.getLogger("matplotlib").setLevel(logging.WARNING) @@ -197,7 +197,9 @@ def log_exception(args: threading.ExceptHookArgs): class ConcurrentRunner: - threads: ClassVar[list[threading.Thread]] = [] + # Sub-tests to run, as (name, target, args) triples. Threads are created by + # run(), so that the pool decides how many exist at once. + tests: ClassVar[list[tuple[str, object, object]]] = [] # Env var to filter sub-tests by exact name match. Value is a # '|'-separated list, e.g. CR_FILTER="testname1|testname2". When set, @@ -232,7 +234,44 @@ 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 caps how many run at once. Half the cores + keeps a 16-core CI runner at roughly the peak load its heaviest test + already sustains today. + """ + cores_count = len(os.sched_getaffinity(0)) + return max(2, cores_count // 2) + + 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 +286,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 966fd17c048d..a71017fdab17 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), ) + # Previously run sequentially at the end of run_all. Each 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 8e4c743cc0fb..c0a676c54615 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,94 @@ 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) + + +def run_recovery_elections(args): + with partitioned_network(args) as network: 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) 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) + # A single sequential network used to dominate this test's run time. The + # groups below each run on their own network, concurrently, and preserve the + # original relative order of the tests they contain. + 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 to not getting CPU promptly: a starved + # node looks like a partitioned one. Run fewer at once than the default. + cr.run(max_concurrent=4) diff --git a/tests/reconfiguration.py b/tests/reconfiguration.py index 8293f6395e11..8f381be9b677 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) + + +# These used to run sequentially at the end of run_all. They each build their +# own single-node network and share no state with it, so they are registered as +# their own sub-tests and run concurrently instead. +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 c32ee890a45c..d2ab423609b2 100644 --- a/tests/schema.py +++ b/tests/schema.py @@ -293,12 +293,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() From c22175b1df189d0126747db0abea567782db1ca2 Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 16:45:09 +0100 Subject: [PATCH 2/6] Fix import order and make partition rule changes atomic Two problems from the previous commit's CI run. Job A failed ruff's import-order check on tests/infra/runner.py and tests/infra/partitions.py. `infra.*` belongs in its own first-party block after the third-party one, as it was before. This was self-inflicted: ruff was run over a directory holding only the changed files, where `infra` does not resolve as first-party, so its autofix merged the blocks and then agreed with itself. Job C failed with `basic` timing out waiting for an election, 75s after dropping its partition, while three other partitioned networks were running. iptc.easy operates on a table object that is shared process-wide and is committed and refreshed on every call, so interleaved calls from several threads lose updates: one thread's refresh can discard another's pending change, leaving stale DROP rules in place. That is exactly what a node which never rejoins an election looks like. The lock was previously taken per rule, which leaves a partition observable half-applied and does nothing about a concurrent refresh landing mid-set. Hold it across the whole set of rules for a partition, an isolation and a drop, so each appears and disappears in one step. Lower the partitions concurrency from 4 to 2 as well, since these groups are the least tolerant in the suite both of iptables contention and of not getting CPU promptly. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 24ad6978-a80f-45bc-b73b-268aeaf2b30e --- tests/infra/partitions.py | 57 ++++++++++++++++++++++++--------------- tests/infra/runner.py | 3 ++- tests/partitions_test.py | 7 ++--- 3 files changed, 41 insertions(+), 26 deletions(-) diff --git a/tests/infra/partitions.py b/tests/infra/partitions.py index 17a64fb0e384..c5c43a11c4dc 100644 --- a/tests/infra/partitions.py +++ b/tests/infra/partitions.py @@ -7,11 +7,12 @@ import threading from dataclasses import field -import infra.network -import infra.node import iptc from loguru import logger as LOG +import infra.network +import infra.node + # 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. @@ -23,11 +24,14 @@ _chain_counter = itertools.count() _chain_counter_lock = threading.Lock() -# libiptc reads the whole table, modifies it and writes it back, so two threads -# committing at once can silently lose each other's rules. Every mutation of the -# filter table goes through this lock, which is enough because all partitioned -# networks in a test run live in one process (infra.runner.ConcurrentRunner -# threads). +# 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() @@ -123,8 +127,11 @@ def drop(self): LOG.info(f'Dropping rules "{self.name or "[unamed]"}"') if self.chain_name is None: return - for rule in self.rules: - _drop_rule(self.chain_name, rule) + # 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: @@ -272,8 +279,11 @@ def isolate_node( if isolation_dir & IsolationDir.OUTBOUND_RESPONSES: rules.append(self.reverse_rule(client_rule)) - for rule in rules: - _replace_rule(self.chain_name, 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) @@ -320,19 +330,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 diff --git a/tests/infra/runner.py b/tests/infra/runner.py index 21ca38d206be..4fe3d85f41ac 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -13,12 +13,13 @@ from typing import ClassVar import better_exceptions +from loguru import logger as LOG + import infra.bencher import infra.jwt_issuer import infra.network import infra.proc import infra.remote_client -from loguru import logger as LOG logging.getLogger("matplotlib").setLevel(logging.WARNING) diff --git a/tests/partitions_test.py b/tests/partitions_test.py index c0a676c54615..9af7c706971d 100644 --- a/tests/partitions_test.py +++ b/tests/partitions_test.py @@ -1638,6 +1638,7 @@ def run_recovery_elections(args): ) # These groups deliberately isolate nodes and wait for elections, so they - # are the most sensitive in the suite to not getting CPU promptly: a starved - # node looks like a partitioned one. Run fewer at once than the default. - cr.run(max_concurrent=4) + # 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) From acefb0c03426b06d2abd6b8e93e769ab265cb159 Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 17:22:31 +0100 Subject: [PATCH 3/6] Raise the default sub-test concurrency to one per core The first measurement of the bounded pool showed recovery_test going from 166.9s to 184.4s. It has nineteen sub-tests and ran unbounded before, so a cap of half the cores was throttling a test that was already comfortable: eight concurrent sub-tests is below what it was doing, not above. The cap exists to bound growth from newly split tests, not to tighten what already worked. Set it to one per core, which leaves recovery_test and the other large pools running as they did, and keep the explicit lower value in partitions_test, whose sub-tests are the ones that actually need it. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 24ad6978-a80f-45bc-b73b-268aeaf2b30e --- tests/infra/runner.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/tests/infra/runner.py b/tests/infra/runner.py index 4fe3d85f41ac..c7a4ccc62477 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -244,12 +244,16 @@ def default_max_concurrent(): 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 caps how many run at once. Half the cores - keeps a 16-core CI runner at roughly the peak load its heaviest test - already sustains today. + and dropped sessions, so this bounds how many run at once. + + One per core. Runners were previously unbounded, and the largest of them + sustained around fifteen concurrent nodes on a sixteen-core CI runner + without trouble, so this is a ceiling on new growth rather than a + tightening of what already worked. 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 // 2) + return max(2, cores_count) def _resolve_max_concurrent(self, max_concurrent): limits = [max_concurrent or self.default_max_concurrent()] From 82bb562629fc9501aaff99001b46568c2a61d567 Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 18:14:41 +0100 Subject: [PATCH 4/6] Stop networks that a test recovers into Two governance_test nodes outlived their test by about 8.5 minutes in CI, logging roughly 660 lines a minute on a join retry loop through the rest of the bucket. `test_all_members` stops the network it is given and recovers into a new one, which it returns. `gov` rebound its local `network` to that, but the enclosing `infra.network.network` context manager still owns the original, so nothing ever stopped the recovered network. The nodes that leaked were the ones `test_no_quote` and `test_node_data` deliberately add without ever trusting, which is why they sat retrying rather than exiting. Give `gov` explicit ownership of the recovered network and stop it in a finally block. The node logs of the leaked nodes were clean, so this reaps them without newly failing on anything. `partitions_test.run_recovery_elections` has the same shape: `test_recovery_elections` stops both the original and its intermediate network but returns a third that nobody stopped. Fix it the same way, which matters more now that the partitions groups run concurrently and a stray node is a neighbour of a live test rather than of an idle runner. The pattern is wider than these two: any test that recovers into a new network and returns it leaks unless its caller takes ownership, and recovery.py does this in several places. Left alone here. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 24ad6978-a80f-45bc-b73b-268aeaf2b30e --- tests/governance.py | 30 ++++++++++++++++++++---------- tests/partitions_test.py | 12 ++++++++++-- 2 files changed, 30 insertions(+), 12 deletions(-) diff --git a/tests/governance.py b/tests/governance.py index f2ab535a914a..e3831cfd118d 100644 --- a/tests/governance.py +++ b/tests/governance.py @@ -568,16 +568,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/partitions_test.py b/tests/partitions_test.py index 9af7c706971d..27cb1396f124 100644 --- a/tests/partitions_test.py +++ b/tests/partitions_test.py @@ -1607,8 +1607,16 @@ def run_forwarding_and_sessions(args): def run_recovery_elections(args): with partitioned_network(args) as network: - network = test_recovery_elections(network, args) - test_ledger_invariants(network, args) + # 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__": From 9494d70c4312c5aca10e0e578394a64c35d894e4 Mon Sep 17 00:00:00 2001 From: achamayou Date: Tue, 1 Sep 2026 18:21:07 +0100 Subject: [PATCH 5/6] Address review: explicit chain-name check, per-instance sub-test list The iptables chain-name limit was enforced with an assert, which python -O strips, turning a violation into an obscure iptables failure later. Raise instead. ConcurrentRunner.tests was a class attribute, so two runners in one process would share accumulated sub-tests. Move it into __init__ so each instance owns its own list. Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 24ad6978-a80f-45bc-b73b-268aeaf2b30e --- tests/infra/partitions.py | 6 +++++- tests/infra/runner.py | 8 ++++---- 2 files changed, 9 insertions(+), 5 deletions(-) diff --git a/tests/infra/partitions.py b/tests/infra/partitions.py index c5c43a11c4dc..8e175f0ea4eb 100644 --- a/tests/infra/partitions.py +++ b/tests/infra/partitions.py @@ -41,7 +41,11 @@ def _next_chain_name(): # 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}" - assert len(name) <= MAX_CHAIN_NAME_LENGTH, name + 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 diff --git a/tests/infra/runner.py b/tests/infra/runner.py index c7a4ccc62477..916b260c5564 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -198,10 +198,6 @@ def log_exception(args: threading.ExceptHookArgs): class ConcurrentRunner: - # Sub-tests to run, as (name, target, args) triples. Threads are created by - # run(), so that the pool decides how many exist at once. - tests: ClassVar[list[tuple[str, object, object]]] = [] - # 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. @@ -227,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: From 4bff67362c9f05cde55aa7b857537c0b7725cd7f Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 4 Sep 2026 08:30:59 +0000 Subject: [PATCH 6/6] Rephrase comments to describe current state, not history Co-authored-by: eddyashton <6000239+eddyashton@users.noreply.github.com> --- tests/e2e_operations.py | 7 +++---- tests/infra/runner.py | 10 +++++----- tests/nodes.py | 4 ++-- tests/partitions_test.py | 5 ++--- tests/reconfiguration.py | 6 +++--- tests/schema.py | 3 +-- 6 files changed, 16 insertions(+), 19 deletions(-) diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index c5190f56ba86..dec9d8f16536 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -3784,10 +3784,9 @@ def _run_backup_snapshot_download(const_args, label_suffix, tests): test(network, args) -# These used to run as one sequential group, which made it the longest sub-test -# in schema_test by a wide margin. Each group below brings up its own network, -# so they run concurrently. Every test starts by finding the primary and issuing -# its own transactions, so none of them depend on the others. +# 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, diff --git a/tests/infra/runner.py b/tests/infra/runner.py index 916b260c5564..be5b53c5e841 100644 --- a/tests/infra/runner.py +++ b/tests/infra/runner.py @@ -246,11 +246,11 @@ def default_max_concurrent(): network that cannot get CPU promptly sees spurious leadership elections and dropped sessions, so this bounds how many run at once. - One per core. Runners were previously unbounded, and the largest of them - sustained around fifteen concurrent nodes on a sixteen-core CI runner - without trouble, so this is a ceiling on new growth rather than a - tightening of what already worked. Tests whose sub-tests are unusually - sensitive, such as partitions_test, pass a lower value to run(). + 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) diff --git a/tests/nodes.py b/tests/nodes.py index a71017fdab17..7b9429d2ba01 100644 --- a/tests/nodes.py +++ b/tests/nodes.py @@ -322,8 +322,8 @@ def add(parser): nodes=infra.e2e_args.min_nodes(cr.args, f=1), ) - # Previously run sequentially at the end of run_all. Each builds its own - # single-node network, so they run concurrently with everything else. + # 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), ( diff --git a/tests/partitions_test.py b/tests/partitions_test.py index 27cb1396f124..e5a0eed44eed 100644 --- a/tests/partitions_test.py +++ b/tests/partitions_test.py @@ -1625,9 +1625,8 @@ def run_recovery_elections(args): 20 # Increase snapshot frequency for faster reconfigurations ) - # A single sequential network used to dominate this test's run time. The - # groups below each run on their own network, concurrently, and preserve the - # original relative order of the tests they contain. + # 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), diff --git a/tests/reconfiguration.py b/tests/reconfiguration.py index 8f381be9b677..452464e5d414 100644 --- a/tests/reconfiguration.py +++ b/tests/reconfiguration.py @@ -1420,9 +1420,9 @@ def run_ipv6(args): run_all(args, ipv6=True) -# These used to run sequentially at the end of run_all. They each build their -# own single-node network and share no state with it, so they are registered as -# their own sub-tests and run concurrently instead. +# 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() diff --git a/tests/schema.py b/tests/schema.py index d2ab423609b2..6793af01fe61 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),