From 7f86311594668a69cddd7f0bed4031b06d8ba99b Mon Sep 17 00:00:00 2001 From: cjen1 Date: Tue, 8 Sep 2026 15:32:36 +0000 Subject: [PATCH 1/5] Separate snapshot inspection from snapshot staging Add node-local committed snapshot listing and passive target-seqno waiting. Migrate timing, election, and persistence callers without changing generation or recovery-copy behavior. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CMakeLists.txt | 6 +++ tests/e2e_operations.py | 101 ++++++++++++++----------------------- tests/infra/node.py | 53 ++++++++++++++++++++ tests/snapshot_files.py | 108 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 204 insertions(+), 64 deletions(-) create mode 100644 tests/snapshot_files.py diff --git a/CMakeLists.txt b/CMakeLists.txt index 6f600245a758..968f6be00dbf 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1066,6 +1066,12 @@ if(BUILD_TESTS) COMMAND ${PYTHON} ${CMAKE_SOURCE_DIR}/tests/infra/github.py ) set_property(TEST github_version_lts_test APPEND PROPERTY LABELS bucket_c) + + add_test( + NAME snapshot_files_test + COMMAND ${PYTHON} ${CMAKE_SOURCE_DIR}/tests/snapshot_files.py + ) + set_property(TEST snapshot_files_test PROPERTY LABELS unit) endif() if(NOT TSAN) diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index e5257f3af493..efd6107d3c05 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -3808,12 +3808,7 @@ def run_propose_request_vote(const_args): original_primary, original_term = network.find_primary() LOG.info("Waiting for initial snapshot") - network.get_committed_snapshots( - original_primary, - target_seqno=1, - force_txs=False, - wait_for_target_seqno=True, - ) + original_primary.wait_for_snapshot(1) network.wait_for_node_commit_sync(timeout=16) original_primary.remote.remote.proc.send_signal(signal.SIGTERM) @@ -3861,29 +3856,11 @@ def net_with_min_tx(label, min_tx_interval): net.start_and_open(inner_args) yield net - def get_committed_snapshot_files(net): - primary, _ = net.find_primary() - snapshots_dirs = [ - os.path.join(primary.remote.remote.root, primary.remote.snapshots_dir_name) - ] - if primary.remote.read_only_snapshots_dir_name is not None: - snapshots_dirs.append( - os.path.join( - primary.remote.remote.root, - primary.remote.read_only_snapshots_dir_name, - ) - ) - - snapshots = set() - for snapshots_dir in snapshots_dirs: - if not os.path.isdir(snapshots_dir): - continue - - for snapshot_name in os.listdir(snapshots_dir): - if ccf.ledger.is_snapshot_file_committed(snapshot_name): - snapshots.add(snapshot_name) - - return snapshots + def snapshot_names(node): + return { + os.path.basename(path) + for path in node.get_snapshots(include_read_only=True) + } # Pattern for these tests: # 1. wait for any startup triggered txs to commit and net to settle @@ -3894,15 +3871,15 @@ def get_committed_snapshot_files(net): def run_low(): with net_with_min_tx("_low", 0) as net: time.sleep(1) - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, - ) - baseline = get_committed_snapshot_files(net) + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] + ) + primary.wait_for_snapshot(target.seqno, timeout=5) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( len(final - baseline) >= 8 ), f"With min_tx_interval set to 0 we expect snapshots to be generated at around 1 per second, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." @@ -3910,18 +3887,18 @@ def run_low(): def run_exact(): with net_with_min_tx("_exact", 2) as net: time.sleep(1) - try: - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] ) + try: + primary.wait_for_snapshot(target.seqno, timeout=5) except TimeoutError: pass - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 2 we expect no snapshots to be generated without transactions, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." @@ -3929,38 +3906,32 @@ def run_exact(): def run_high(): with net_with_min_tx("_high", 10) as net: time.sleep(1) - try: - net.get_committed_snapshots( - net.find_primary()[0], - force_txs=False, - wait_for_target_seqno=True, - timeout=5, + primary, _ = net.find_primary() + with primary.client() as c: + target = TxID.from_str( + c.get("/node/commit").body.json()["transaction_id"] ) + try: + primary.wait_for_snapshot(target.seqno, timeout=5) except TimeoutError: pass - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 10 we expect no snapshots to be generated without transactions, but got {final} snapshots 10s after a baseline of {baseline}, with {final - baseline} new snapshots seen over the test." tx_id = net.txs.issue(net, number_txs=1) - baseline = get_committed_snapshot_files(net) + baseline = snapshot_names(primary) time.sleep(10) - final = get_committed_snapshot_files(net) + final = snapshot_names(primary) assert ( final == baseline ), f"With min_tx_interval set to 10 and we expect no snapshots to be generated with only one extra tx, but got {final} snapshots 10s after a baseline of {baseline}, and in total saw {final - baseline} new snapshots over the test." net.txs.issue(net, number_txs=20) - primary, _ = net.find_primary() - net.get_committed_snapshots( - primary, - target_seqno=tx_id.seqno, - force_txs=False, - wait_for_target_seqno=True, - ) + primary.wait_for_snapshot(tx_id.seqno) with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: futures = [ @@ -4015,8 +3986,10 @@ def run_snapshot_persistence_across_primary_failure(const_args): elapsed = time.time() - start_time snapshots = set() for node in net.nodes: - snapshots_dir = net.get_committed_snapshots(node, force_txs=False) - snapshots = snapshots.union(set(os.listdir(snapshots_dir))) + snapshots.update( + os.path.basename(path) + for path in node.get_snapshots(include_read_only=True) + ) total_snapshots = len(snapshots) diff --git a/tests/infra/node.py b/tests/infra/node.py index 374d73f8f942..a78706e7e016 100644 --- a/tests/infra/node.py +++ b/tests/infra/node.py @@ -864,6 +864,59 @@ def get_ledger(self): return current_ledger_dir, [committed_ledger_dir] + def get_snapshots(self, *, include_read_only=False) -> list[str]: + """List committed snapshot paths on this node, ordered by snapshot seqno. + + Paths are node-owned: copy them before modifying them or relying on them + surviving cleanup. Read-only startup snapshots are excluded by default. + """ + directories = [self.remote.snapshots_dir_name] + if include_read_only and self.remote.read_only_snapshots_dir_name is not None: + directories.append(self.remote.read_only_snapshots_dir_name) + + snapshots = [] + for directory in directories: + path = os.path.join(self.remote.remote.root, directory) + try: + with os.scandir(path) as entries: + snapshots.extend( + entry.path + for entry in entries + if entry.name.startswith("snapshot_") + and ccf.ledger.is_snapshot_file_committed(entry.name) + and entry.is_file() + ) + except FileNotFoundError: + LOG.debug(f"Snapshot directory does not exist yet: {path}") + + return sorted(snapshots, key=ccf.ledger.snapshot_index_from_filename) + + def wait_for_snapshot(self, target_seqno, timeout=20) -> str: + """Wait for a committed snapshot in this node's writable directory. + + The snapshot state must include target_seqno. This does not emit + transactions, trigger snapshots, or copy files. + """ + LOG.info( + f"Waiting for node {self.local_node_id} snapshot including seqno {target_seqno}" + ) + end_time = time.monotonic() + timeout + while True: + snapshots = self.get_snapshots() + for snapshot in snapshots: + if ccf.ledger.snapshot_index_from_filename(snapshot)[0] >= target_seqno: + LOG.info(f"Found committed snapshot {snapshot}") + return snapshot + + remaining = end_time - time.monotonic() + if remaining <= 0: + raise TimeoutError( + f"Could not find committed snapshot on node {self.local_node_id} " + f"including seqno {target_seqno} after {timeout}s; " + f"snapshot files: {snapshots}" + ) + time.sleep(min(0.1, remaining)) + def get_committed_snapshots(self, pre_condition_func=lambda src_dir, _: True): ( main_snapshots_dir, diff --git a/tests/snapshot_files.py b/tests/snapshot_files.py new file mode 100644 index 000000000000..25df0d762728 --- /dev/null +++ b/tests/snapshot_files.py @@ -0,0 +1,108 @@ +# Copyright (c) Microsoft Corporation. All rights reserved. +# Licensed under the Apache 2.0 License. + +import tempfile +import unittest +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock, patch + +import infra.interfaces +from infra.node import Node + + +class SnapshotFilesTest(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.root = Path(self.directory.name) + self.writable = self.root / "snapshots" + self.read_only = self.root / "snapshots.ro" + self.writable.mkdir() + self.read_only.mkdir() + self.node = Node(0, infra.interfaces.HostSpec(rpc_interfaces={})) + self.node.remote = SimpleNamespace( + remote=SimpleNamespace(root=str(self.root)), + snapshots_dir_name=self.writable.name, + read_only_snapshots_dir_name=self.read_only.name, + ) + self.node.common_dir = str(self.root / "common") + self.node.client = Mock(side_effect=AssertionError("Unexpected HTTP request")) + + def tearDown(self): + self.node.client.assert_not_called() + self.assertFalse(Path(self.node.common_dir).exists()) + + def snapshot(self, name, directory=None): + path = (directory if directory is not None else self.writable) / name + path.write_bytes(b"snapshot") + return str(path) + + def test_committed_files_in_sequence_order(self): + newer = self.snapshot("snapshot_100_101.committed") + older = self.snapshot("snapshot_20_21.committed") + self.snapshot("snapshot_200_201") + self.snapshot("snapshot_300_301.committed.ignored") + self.snapshot("not_a_snapshot.committed") + (self.writable / "snapshot_400_401.committed").mkdir() + self.assertEqual(self.node.get_snapshots(), [older, newer]) + + def test_read_only_is_explicit_and_preserves_paths(self): + name = "snapshot_20_21.committed" + writable = self.snapshot(name) + read_only = self.snapshot(name, self.read_only) + self.assertEqual(self.node.get_snapshots(), [writable]) + self.assertEqual( + self.node.get_snapshots(include_read_only=True), [writable, read_only] + ) + self.node.remote.read_only_snapshots_dir_name = None + self.assertEqual(self.node.get_snapshots(include_read_only=True), [writable]) + + def test_missing_directory_is_empty(self): + self.writable.rmdir() + self.assertEqual(self.node.get_snapshots(), []) + + def test_wait_uses_snapshot_state_not_evidence(self): + self.snapshot("snapshot_20_100.committed") + target = self.snapshot("snapshot_30_101.committed") + self.assertEqual(self.node.wait_for_snapshot(30, timeout=0), target) + with self.assertRaises(TimeoutError): + self.node.wait_for_snapshot(31, timeout=0) + + def test_wait_is_pinned_to_writable_directory(self): + self.snapshot("snapshot_100_101.committed", self.read_only) + other_node = self.root / "other_node" + other_node.mkdir() + self.snapshot("snapshot_100_101.committed", other_node) + with self.assertRaisesRegex(TimeoutError, "node 0.*seqno 100"): + self.node.wait_for_snapshot(100, timeout=0) + + def test_wait_observes_commit_rename(self): + pending = Path(self.snapshot("snapshot_20_21")) + committed = pending.with_name(pending.name + ".committed") + with patch( + "infra.node.time.sleep", side_effect=lambda _: pending.rename(committed) + ) as sleep: + self.assertEqual(self.node.wait_for_snapshot(20, timeout=1), str(committed)) + sleep.assert_called_once() + + def test_filesystem_errors_propagate(self): + with ( + patch("infra.node.os.scandir", side_effect=PermissionError("denied")), + self.assertRaises(PermissionError), + ): + self.node.get_snapshots() + + def test_wait_timeout_is_not_restarted(self): + with ( + patch("infra.node.time.monotonic", side_effect=[0, 0.9, 1]), + patch("infra.node.time.sleep") as sleep, + self.assertRaises(TimeoutError), + ): + self.node.wait_for_snapshot(20, timeout=1) + sleep.assert_called_once() + self.assertAlmostEqual(sleep.call_args.args[0], 0.1) + + +if __name__ == "__main__": + unittest.main() From 1d6474b5a2e0ffaba0aed4e75a9ca54b36913422 Mon Sep 17 00:00:00 2001 From: cjen1 Date: Tue, 8 Sep 2026 15:42:15 +0000 Subject: [PATCH 2/5] Use existing end-to-end coverage for snapshot helpers Remove the helper-only snapshot file tests and their CMake registration. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- CMakeLists.txt | 6 --- tests/snapshot_files.py | 108 ---------------------------------------- 2 files changed, 114 deletions(-) delete mode 100644 tests/snapshot_files.py diff --git a/CMakeLists.txt b/CMakeLists.txt index ba6d8a811b42..814dd9e44847 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -1083,12 +1083,6 @@ if(BUILD_TESTS) COMMAND ${PYTHON} ${CMAKE_SOURCE_DIR}/tests/infra/github.py ) set_property(TEST github_version_lts_test APPEND PROPERTY LABELS bucket_c) - - add_test( - NAME snapshot_files_test - COMMAND ${PYTHON} ${CMAKE_SOURCE_DIR}/tests/snapshot_files.py - ) - set_property(TEST snapshot_files_test PROPERTY LABELS unit) endif() if(NOT TSAN) diff --git a/tests/snapshot_files.py b/tests/snapshot_files.py deleted file mode 100644 index 25df0d762728..000000000000 --- a/tests/snapshot_files.py +++ /dev/null @@ -1,108 +0,0 @@ -# Copyright (c) Microsoft Corporation. All rights reserved. -# Licensed under the Apache 2.0 License. - -import tempfile -import unittest -from pathlib import Path -from types import SimpleNamespace -from unittest.mock import Mock, patch - -import infra.interfaces -from infra.node import Node - - -class SnapshotFilesTest(unittest.TestCase): - def setUp(self): - self.directory = tempfile.TemporaryDirectory() - self.addCleanup(self.directory.cleanup) - self.root = Path(self.directory.name) - self.writable = self.root / "snapshots" - self.read_only = self.root / "snapshots.ro" - self.writable.mkdir() - self.read_only.mkdir() - self.node = Node(0, infra.interfaces.HostSpec(rpc_interfaces={})) - self.node.remote = SimpleNamespace( - remote=SimpleNamespace(root=str(self.root)), - snapshots_dir_name=self.writable.name, - read_only_snapshots_dir_name=self.read_only.name, - ) - self.node.common_dir = str(self.root / "common") - self.node.client = Mock(side_effect=AssertionError("Unexpected HTTP request")) - - def tearDown(self): - self.node.client.assert_not_called() - self.assertFalse(Path(self.node.common_dir).exists()) - - def snapshot(self, name, directory=None): - path = (directory if directory is not None else self.writable) / name - path.write_bytes(b"snapshot") - return str(path) - - def test_committed_files_in_sequence_order(self): - newer = self.snapshot("snapshot_100_101.committed") - older = self.snapshot("snapshot_20_21.committed") - self.snapshot("snapshot_200_201") - self.snapshot("snapshot_300_301.committed.ignored") - self.snapshot("not_a_snapshot.committed") - (self.writable / "snapshot_400_401.committed").mkdir() - self.assertEqual(self.node.get_snapshots(), [older, newer]) - - def test_read_only_is_explicit_and_preserves_paths(self): - name = "snapshot_20_21.committed" - writable = self.snapshot(name) - read_only = self.snapshot(name, self.read_only) - self.assertEqual(self.node.get_snapshots(), [writable]) - self.assertEqual( - self.node.get_snapshots(include_read_only=True), [writable, read_only] - ) - self.node.remote.read_only_snapshots_dir_name = None - self.assertEqual(self.node.get_snapshots(include_read_only=True), [writable]) - - def test_missing_directory_is_empty(self): - self.writable.rmdir() - self.assertEqual(self.node.get_snapshots(), []) - - def test_wait_uses_snapshot_state_not_evidence(self): - self.snapshot("snapshot_20_100.committed") - target = self.snapshot("snapshot_30_101.committed") - self.assertEqual(self.node.wait_for_snapshot(30, timeout=0), target) - with self.assertRaises(TimeoutError): - self.node.wait_for_snapshot(31, timeout=0) - - def test_wait_is_pinned_to_writable_directory(self): - self.snapshot("snapshot_100_101.committed", self.read_only) - other_node = self.root / "other_node" - other_node.mkdir() - self.snapshot("snapshot_100_101.committed", other_node) - with self.assertRaisesRegex(TimeoutError, "node 0.*seqno 100"): - self.node.wait_for_snapshot(100, timeout=0) - - def test_wait_observes_commit_rename(self): - pending = Path(self.snapshot("snapshot_20_21")) - committed = pending.with_name(pending.name + ".committed") - with patch( - "infra.node.time.sleep", side_effect=lambda _: pending.rename(committed) - ) as sleep: - self.assertEqual(self.node.wait_for_snapshot(20, timeout=1), str(committed)) - sleep.assert_called_once() - - def test_filesystem_errors_propagate(self): - with ( - patch("infra.node.os.scandir", side_effect=PermissionError("denied")), - self.assertRaises(PermissionError), - ): - self.node.get_snapshots() - - def test_wait_timeout_is_not_restarted(self): - with ( - patch("infra.node.time.monotonic", side_effect=[0, 0.9, 1]), - patch("infra.node.time.sleep") as sleep, - self.assertRaises(TimeoutError), - ): - self.node.wait_for_snapshot(20, timeout=1) - sleep.assert_called_once() - self.assertAlmostEqual(sleep.call_args.args[0], 0.1) - - -if __name__ == "__main__": - unittest.main() From e328f280b2602b84778849daf69035307d00205f Mon Sep 17 00:00:00 2001 From: cjen1 Date: Tue, 8 Sep 2026 15:45:36 +0000 Subject: [PATCH 3/5] Read governance and recovery snapshots directly from disk Use the existing snapshot trigger's target when reading governance history. Compare recovery snapshot files in place, including read-only inputs, without copying into common_dir. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/governance_history.py | 5 ++--- tests/recovery_snapshot_endorsements.py | 27 ++++++++++++++++--------- 2 files changed, 19 insertions(+), 13 deletions(-) diff --git a/tests/governance_history.py b/tests/governance_history.py index 13787f0da8b4..6d9a1edef31f 100644 --- a/tests/governance_history.py +++ b/tests/governance_history.py @@ -4,7 +4,6 @@ import base64 import http import json -import os import ccf.ledger import ccf.read_ledger @@ -250,9 +249,9 @@ def fmt_str(data: bytes) -> str: tables_format_rules=format_rule, ) - snapshot_dir = network.get_committed_snapshots(primary) + snapshot_path = primary.wait_for_snapshot(target_seqno) assert ccf.read_ledger.run( - paths=[os.path.join(snapshot_dir, os.listdir(snapshot_dir)[-1])], + paths=[snapshot_path], print_mode=ccf.read_ledger.PrintMode.Contents, is_snapshot=True, tables_format_rules=format_rule, diff --git a/tests/recovery_snapshot_endorsements.py b/tests/recovery_snapshot_endorsements.py index e3db52595c44..64c2b18ebe97 100644 --- a/tests/recovery_snapshot_endorsements.py +++ b/tests/recovery_snapshot_endorsements.py @@ -111,14 +111,21 @@ def _copy_ledger_prefix(source_dirs, destination, first_excluded_seqno): assert copied > 0 -def _assert_node_snapshot_unchanged( - network, node, snapshot_name, expected_snapshot_digest -): - snapshots_dir = network.get_committed_snapshots(node, force_txs=False) - snapshot_path = os.path.join(snapshots_dir, snapshot_name) - assert os.path.isfile(snapshot_path), snapshot_path - with open(snapshot_path, "rb") as snapshot_file: - assert hashlib.sha256(snapshot_file.read()).digest() == expected_snapshot_digest +def _assert_node_snapshot_unchanged(node, snapshot_name, expected_snapshot_digest): + snapshot_paths = [ + path + for path in node.get_snapshots(include_read_only=True) + if os.path.basename(path) == snapshot_name + ] + assert ( + snapshot_paths + ), f"Snapshot {snapshot_name} not found on node {node.local_node_id}" + for snapshot_path in snapshot_paths: + with open(snapshot_path, "rb") as snapshot_file: + assert ( + hashlib.sha256(snapshot_file.read()).digest() + == expected_snapshot_digest + ), snapshot_path def run_recovery_snapshot_endorsements(args): @@ -237,7 +244,7 @@ def run_recovery_snapshot_endorsements(args): < logs.index(public_recovery_log) ) _assert_node_snapshot_unchanged( - valid_attempt, valid_primary, snapshot_name, snapshot_digest + valid_primary, snapshot_name, snapshot_digest ) finally: _stop_incomplete_recovery(valid_attempt) @@ -272,7 +279,7 @@ def run_recovery_snapshot_endorsements(args): assert "No usable local snapshot found" in logs assert "Setting startup snapshot seqno" not in logs _assert_node_snapshot_unchanged( - fallback_attempt, fallback_primary, snapshot_name, snapshot_digest + fallback_primary, snapshot_name, snapshot_digest ) finally: _stop_incomplete_recovery(fallback_attempt) From 639b2491976e7cfad22de3d2b057c4c598f46500 Mon Sep 17 00:00:00 2001 From: cjen1 Date: Wed, 9 Sep 2026 13:31:00 +0000 Subject: [PATCH 4/5] Make snapshot generation explicit in test setup Replace eight combined snapshot-helper calls with explicit triggers and passive waits. Use committed application transactions as snapshot targets and read generated files directly. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/e2e_operations.py | 78 +++++++++---------------- tests/recovery_snapshot_endorsements.py | 10 +--- 2 files changed, 30 insertions(+), 58 deletions(-) diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index 67006f397322..3abff9f9a32f 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -176,21 +176,6 @@ def test_forced_ledger_chunk(network, args): return network -def find_snapshot_after_seqno(snapshots_dir, seqno): - for snapshot_name in os.listdir(snapshots_dir): - with ccf.ledger.Snapshot( - os.path.join(snapshots_dir, snapshot_name) - ) as snapshot: - snapshot_seqno = snapshot.get_public_domain().get_seqno() - if snapshot_seqno > seqno: - LOG.info(f"Found a snapshot at {snapshot_seqno} which is after {seqno}") - return snapshot_seqno - - raise RuntimeError( - f"Could not find a snapshot after seqno {seqno} in {snapshots_dir}" - ) - - def find_latest_committed_snapshot_name(network, count=1): assert count > 0, f"Expected positive snapshot count, got {count}" primary, _ = network.find_primary() @@ -231,10 +216,9 @@ def test_forced_snapshot(network, args): # Issue some more transactions network.txs.issue(network, number_txs=5) - snapshots_dir = network.get_committed_snapshots( - primary, target_seqno=hwm_pre_proposal + 1, wait_for_target_seqno=True - ) - find_snapshot_after_seqno(snapshots_dir, hwm_pre_proposal) + snapshot_path = primary.wait_for_snapshot(hwm_pre_proposal + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_public_domain().get_seqno() > hwm_pre_proposal # Do not issue another transaction after this call. The snapshot request # must make all preceding transactions available in a committed chunk even @@ -283,13 +267,10 @@ def issue_governance_txs(count): issue_governance_txs(5) - snapshots_dir = network.get_committed_snapshots( - primary, - target_seqno=hwm_pre_proposal + 1, - force_txs=False, - wait_for_target_seqno=True, - ) - snapshot_seqno = find_snapshot_after_seqno(snapshots_dir, hwm_pre_proposal) + snapshot_path = primary.wait_for_snapshot(hwm_pre_proposal + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + snapshot_seqno = snapshot.get_public_domain().get_seqno() + assert snapshot_seqno > hwm_pre_proposal _, committed_ledger_dirs = primary.get_ledger() ledger = ccf.ledger.Ledger( @@ -334,13 +315,9 @@ def test_snapshot_create_endpoint(network, args): r = c.post("/node/snapshot:create") assert r.status_code == http.HTTPStatus.NO_CONTENT, r - snapshots_dir = network.get_committed_snapshots( - primary, - target_seqno=hwm_pre_request + 1, - force_txs=False, - wait_for_target_seqno=True, - ) - find_snapshot_after_seqno(snapshots_dir, hwm_pre_request) + snapshot_path = primary.wait_for_snapshot(hwm_pre_request + 1) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_public_domain().get_seqno() > hwm_pre_request return network @@ -1569,8 +1546,10 @@ def test_ledger_chunk_redirect_gap(network, args): commit_seqno = TxID.from_str(r["transaction_id"]).seqno new_node = network.create_node() - # force primary to generate a new snapshot after commit idx - network.get_committed_snapshots() + # Commit a transaction beyond the old boundary before requesting a snapshot. + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) network.join_node( new_node, args.package, @@ -3455,8 +3434,9 @@ def test_join_time_snapshot_fetch_failure(network, args): # Ensure at least one committed snapshot exists so that joining nodes # can be given one (startup_seqno > 0). - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) - network.get_committed_snapshots(primary) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) # Full reconfigure so every remaining node has startup_seqno > 0 # (otherwise a redirect to the primary would let the joiner succeed). @@ -3545,11 +3525,12 @@ def test_error_message_on_failure_to_fetch_snapshot(network, args): ) network.trust_node(new_node, args) - # Issue enough transactions to trigger a new snapshot on the primary. + # Explicitly trigger a snapshot after the new node has joined. # The snapshot_evidence hook on new_node then schedules BackupSnapshotFetch, # which exhausts its 3 attempts (all HTTP 404) and logs "giving up". - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) - network.get_committed_snapshots(primary) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) _assert_snapshot_fetch_failure_messages(new_node, timeout_s=30) @@ -3559,26 +3540,21 @@ def test_backup_snapshot_fetch(network, args): backups = network.find_backups() assert len(backups) > 0, "Expected at least one backup node" - # Issue enough transactions to trigger snapshot generation - # The primary will create a snapshot after snapshot_tx_interval txs - LOG.info("Issuing transactions to trigger snapshot generation") - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() # Wait for committed snapshots on the primary, and use those as expected # snapshot files on backups. LOG.info("Waiting for committed snapshot on primary") - primary_snapshots_dir = network.get_committed_snapshots(primary) + primary.wait_for_snapshot(target.seqno) expected_snapshot_sizes = { - snapshot_name: os.path.getsize( - os.path.join(primary_snapshots_dir, snapshot_name) - ) - for snapshot_name in os.listdir(primary_snapshots_dir) - if ccf.ledger.is_snapshot_file_committed(snapshot_name) + os.path.basename(path): os.path.getsize(path) + for path in primary.get_snapshots(include_read_only=True) } assert ( len(expected_snapshot_sizes) > 0 - ), f"No committed snapshots found in {primary_snapshots_dir}" + ), f"No committed snapshots found on primary {primary.local_node_id}" for backup in backups: backup_snapshots_dir = os.path.join( diff --git a/tests/recovery_snapshot_endorsements.py b/tests/recovery_snapshot_endorsements.py index 64c2b18ebe97..6e604db60b1c 100644 --- a/tests/recovery_snapshot_endorsements.py +++ b/tests/recovery_snapshot_endorsements.py @@ -138,19 +138,15 @@ def run_recovery_snapshot_endorsements(args): initial_network.start_and_open(args) primary, _ = initial_network.find_primary() - app.LoggingTxs("user0").issue( + target = app.LoggingTxs("user0").issue( initial_network, number_txs=2, send_private=False, send_public=True, wait_for_sync=True, ) - snapshot_trigger = primary.trigger_snapshot() - initial_network.get_committed_snapshots( - primary, - target_seqno=snapshot_trigger.seqno, - wait_for_target_seqno=True, - ) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) app.LoggingTxs("user0").issue( initial_network, number_txs=2, From 560baf5c3e74c41c5f28cf40877cebaa848f71e1 Mon Sep 17 00:00:00 2001 From: cjen1 Date: Thu, 10 Sep 2026 15:15:44 +0000 Subject: [PATCH 5/5] Migrate remaining snapshot generation scenarios to disk helpers Use explicit snapshot triggers and passive waits for large snapshots, access and digest checks, selection, and fetch-size limits. Preserve intentional backup copies and assert the oversized snapshot precondition. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- tests/e2e_operations.py | 68 +++++++++++++++++++---------------------- 1 file changed, 31 insertions(+), 37 deletions(-) diff --git a/tests/e2e_operations.py b/tests/e2e_operations.py index 3abff9f9a32f..2e37e1b68536 100644 --- a/tests/e2e_operations.py +++ b/tests/e2e_operations.py @@ -339,36 +339,34 @@ def test_large_snapshot(network, args): log_capture=[], ) - # Force a snapshot at the following signature + target = network.txs.issue(network, number_txs=1) + # Force a snapshot covering the large entries at the following signature. primary.trigger_snapshot() # Check that there is at least a snapshot larger than args.max_msg_size_bytes - snapshots_dir = network.get_committed_snapshots(primary) + snapshot_path = primary.wait_for_snapshot(target.seqno) extra_data_size_bytes = 10000 # Upper bound on additional snapshot data (e.g. receipt) that is passed separately from the snapshot - for s in os.listdir(snapshots_dir): - snapshot_size = os.stat(os.path.join(snapshots_dir, s)).st_size - if snapshot_size > int(args.max_msg_size_bytes) + extra_data_size_bytes: - # Make sure that large snapshot can be parsed - snapshot = ccf.ledger.Snapshot(os.path.join(snapshots_dir, s)) - assert snapshot.get_len() == snapshot_size - LOG.info( - f"Found snapshot [{snapshot_size}] larger than ring buffer max msg size {args.max_msg_size_bytes}" - ) - return network - - raise RuntimeError( - f"Could not find any snapshot file larger than {args.max_msg_size_bytes}" + snapshot_size = os.path.getsize(snapshot_path) + assert snapshot_size > int(args.max_msg_size_bytes) + extra_data_size_bytes, ( + f"Snapshot {snapshot_path} has size {snapshot_size}, expected more than " + f"{int(args.max_msg_size_bytes) + extra_data_size_bytes}" ) + with ccf.ledger.Snapshot(snapshot_path) as snapshot: + assert snapshot.get_len() == snapshot_size + return network def test_snapshot_access(network, args): primary, backups = network.find_nodes() - snapshots_dir = network.get_committed_snapshots(primary) - snapshot_name = ccf.ledger.latest_snapshot(snapshots_dir) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) + snapshot_path = primary.get_snapshots()[-1] + snapshot_name = os.path.basename(snapshot_path) snapshot_index, _ = ccf.ledger.snapshot_index_from_filename(snapshot_name) - with open(os.path.join(snapshots_dir, snapshot_name), "rb") as f: + with open(snapshot_path, "rb") as f: snapshot_data = f.read() for node in (primary, *backups): @@ -535,9 +533,11 @@ def test_snapshot_repr_digest(network, args): """ primary, _ = network.find_nodes() - snapshots_dir = network.get_committed_snapshots(primary) - snapshot_name = ccf.ledger.latest_snapshot(snapshots_dir) - snapshot_path = os.path.join(snapshots_dir, snapshot_name) + target = network.txs.issue(network, number_txs=1) + primary.trigger_snapshot() + primary.wait_for_snapshot(target.seqno) + snapshot_path = primary.get_snapshots()[-1] + snapshot_name = os.path.basename(snapshot_path) with open(snapshot_path, "rb") as f: snapshot_data = f.read() @@ -674,24 +674,15 @@ def test_snapshot_selection(network, args): LOG.info("Creating snapshots") primary, backups = network.find_nodes() - for i in range(3): + for _ in range(max(3, len(backups))): + target = network.txs.issue(network, number_txs=1) primary.trigger_snapshot() - # Snapshot creation and commit takes time. All of the helpers we have to track/poll this - # are expensive, so try a short sleep - time.sleep(1) - - snapshots_dir = network.get_committed_snapshots( - primary, - force_txs=False, - ) + primary.wait_for_snapshot(target.seqno) src_snapshots = [] - for snapshot_name in os.listdir(snapshots_dir): - if ccf.ledger.is_snapshot_file_committed(snapshot_name): - seqno, _ = ccf.ledger.snapshot_index_from_filename(snapshot_name) - src_snapshots.append( - (seqno, snapshot_name, os.path.join(snapshots_dir, snapshot_name)) - ) + for snapshot_path in primary.get_snapshots(): + seqno, _ = ccf.ledger.snapshot_index_from_filename(snapshot_path) + src_snapshots.append((seqno, os.path.basename(snapshot_path), snapshot_path)) src_snapshots.sort() best_snapshot = src_snapshots[-1][1] @@ -3642,7 +3633,10 @@ def assert_no_snapshot_is_present(duration_s=10): ), f"Expected snapshot directory {snapshot_dir} to exist" assert_no_snapshot_is_present() - network.txs.issue(network, number_txs=args.snapshot_tx_interval * 2) + target = network.txs.issue(network, number_txs=1, msg="X" * 2048) + primary.trigger_snapshot() + snapshot_path = primary.wait_for_snapshot(target.seqno) + assert os.path.getsize(snapshot_path) > 1024, snapshot_path assert_no_snapshot_is_present() expected_log_message = "Failed writing received data to disk/application" out_path, _ = new_node.get_logs()