Add node-local snapshot inspection to test infrastructure

tests/e2e_operations.py

@@ -3863,12 +3863,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)
@@ -3916,29 +3911,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
@@ -3949,73 +3926,67 @@ 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."
 
     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."
 
     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 = [
@@ -4070,8 +4041,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)
 

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,

tests/infra/node.py

@@ -877,6 +877,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,

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)