Flaky test fix: Reduce unnecessary reads from live ledgers, and log a warning when they do occur

tests/e2e_operations.py

@@ -155,16 +155,14 @@ def test_forced_ledger_chunk(network, args):
     network.txs.issue(network, number_txs=3)
 
     # Submit a proposal to force a ledger chunk at the following signature
-    proposal = network.consortium.force_ledger_chunk(primary)
+    proposal, _ = network.create_and_wait_for_chunk(primary)
 
     # Issue some more transactions
     network.txs.issue(network, number_txs=5)
 
-    ledger_dirs = primary.remote.ledger_paths()
-
     # Check that there is indeed a ledger chunk that ends at the
     # first signature after proposal.completed_seqno
-    ledger = ccf.ledger.Ledger(ledger_dirs, contiguous_suffix=True)
+    ledger = primary.get_committed_ledger()
     chunk, _, last, next_signature = find_ledger_chunk_for_seqno(
         ledger, proposal.completed_seqno
     )

tests/governance_history.py

@@ -206,8 +206,8 @@ def remove_prefix(s, prefix):
 @reqs.description("Check tables are documented")
 def test_tables_doc(network, args):
     primary, _ = network.find_primary()
-    ledger_directories = primary.remote.ledger_paths()
-    ledger = ccf.ledger.Ledger(ledger_directories, contiguous_suffix=True)
+    network.create_and_wait_for_chunk(primary)
+    ledger = primary.get_committed_ledger()
     table_names_in_ledger = ledger.get_latest_public_state()[0].keys()
     check_all_tables_are_documented(
         table_names_in_ledger, "../doc/audit/builtin_maps.rst"
@@ -219,9 +219,11 @@ def test_tables_doc(network, args):
 def test_ledger_is_readable(network, args):
     primary, backups = network.find_nodes()
     for node in (primary, *backups):
-        ledger_dirs = node.remote.ledger_paths()
+        ledger_dirs = node.get_committed_ledger_dirs()
         LOG.info(f"Reading ledger from {ledger_dirs}")
-        ledger = ccf.ledger.Ledger(ledger_dirs, contiguous_suffix=True)
+        ledger = ccf.ledger.Ledger(
+            ledger_dirs, committed_only=True, contiguous_suffix=True
+        )
         for chunk in ledger:
             for _ in chunk:
                 pass
@@ -236,11 +238,11 @@ def fmt_str(data: bytes) -> str:
     format_rule = [(".*records.*", {"key": fmt_str, "value": fmt_str})]
 
     network.txs.issue(network, number_txs=args.snapshot_tx_interval)
-    network.get_latest_ledger_public_state()
+    network.create_and_wait_for_chunk()
 
     primary, backups = network.find_nodes()
     for node in (primary, *backups):
-        ledger_dirs = node.remote.ledger_paths()
+        ledger_dirs = node.get_committed_ledger_dirs()
         assert ccf.read_ledger.run(
             paths=ledger_dirs,
             print_mode=ccf.read_ledger.PrintMode.Contents,
@@ -274,7 +276,6 @@ def run(args):
 
         network.consortium.set_authenticate_session(args.authenticate_session)
 
-        ledger_directories = primary.remote.ledger_paths()
         LOG.info("Add new member proposal (implicit vote)")
         (
             new_member_proposal,
@@ -323,10 +324,9 @@ def run(args):
             (new_member_proposal.proposal_id, member.service_id, "withdraw")
         )
 
-        # Force ledger flush of all transactions so far
-        network.get_latest_ledger_public_state()
+        network.create_and_wait_for_chunk(primary)
 
-        ledger = ccf.ledger.Ledger(ledger_directories, contiguous_suffix=True)
+        ledger = primary.get_committed_ledger()
         check_operations(ledger, governance_operations)
         check_signatures(ledger)
 

tests/infra/network.py

@@ -2212,23 +2212,54 @@ def wait_for_snapshots_to_be_committed(
 
         return node.get_committed_snapshots(wait_for_snapshots_to_be_committed)
 
-    def _get_ledger_public_view_at(self, node, call, seqno, timeout):
+    def create_and_wait_for_chunk(self, node=None, seqno=None, timeout=5):
+        if node is None:
+            node, _ = self.find_primary()
+
+        proposal = self.consortium.force_ledger_chunk(node)
+        target_seqno = seqno or proposal.completed_seqno
         end_time = time.time() + timeout
-        self.consortium.force_ledger_chunk(node)
+        last_error = None
         while time.time() < end_time:
+            chunk = node.find_committed_ledger_chunk_covering(target_seqno)
+            if chunk is not None:
+                LOG.info(
+                    "Found committed ledger chunk {}-{} covering seqno {} at {}",
+                    chunk[0],
+                    chunk[1],
+                    target_seqno,
+                    chunk[2],
+                )
+                return proposal, chunk
+
             try:
-                return call(seqno)
+                member = self.consortium.get_any_active_member()
+                r = member.update_ack_state_digest(node)
+                with node.client() as c:
+                    c.wait_for_commit(r)
             except Exception as ex:
-                LOG.info(f"Exception: {ex}")
-                time.sleep(0.1)
+                last_error = ex
+                LOG.info(f"Exception while waiting for committed chunk: {ex}")
+            time.sleep(0.1)
+
         raise TimeoutError(
-            f"Could not read transaction at seqno {seqno} from ledger {node.remote.ledger_paths()} after {timeout}s"
+            f"Could not find committed chunk covering seqno {target_seqno} "
+            f"on node {node.local_node_id} after {timeout}s: {last_error}"
         )
 
+    def _get_ledger_public_view_at(self, node, call, seqno, timeout):
+        self.create_and_wait_for_chunk(node=node, seqno=seqno, timeout=timeout)
+        return call(node.get_committed_ledger(), seqno)
+
     def get_ledger_public_state_at(self, seqno, timeout=5):
         primary, _ = self.find_primary()
         return self._get_ledger_public_view_at(
-            primary, primary.get_ledger_public_tables_at, seqno, timeout
+            primary,
+            lambda ledger, s: ledger.get_transaction(s)
+            .get_public_domain()
+            .get_tables(),
+            seqno,
+            timeout,
         )
 
     def get_latest_ledger_public_state(self, timeout=5):
@@ -2238,7 +2269,10 @@ def get_latest_ledger_public_state(self, timeout=5):
             body = resp.body.json()
             tx_id = TxID.from_str(body["transaction_id"])
         return self._get_ledger_public_view_at(
-            primary, primary.get_ledger_public_state_at, tx_id.seqno, timeout
+            primary,
+            lambda ledger, _: ledger.get_latest_public_state(),
+            tx_id.seqno,
+            timeout,
         )
 
     @functools.cached_property

tests/infra/node.py

@@ -547,13 +547,61 @@ def wait_for_node_to_join(self, *args, timeout=3, **kwargs):
 
         raise TimeoutError(f"Node {self.local_node_id} failed to join the network")
 
+    def committed_ledger_chunks(self):
+        chunks = []
+        for ledger_dir in self.remote.ledger_paths():
+            if not os.path.isdir(ledger_dir):
+                continue
+
+            for file_name in os.listdir(ledger_dir):
+                if not ccf.ledger.is_ledger_chunk_committed(file_name):
+                    continue
+                try:
+                    start_seqno, end_seqno = ccf.ledger.range_from_filename(file_name)
+                except ValueError:
+                    continue
+                chunks.append(
+                    (start_seqno, end_seqno, os.path.join(ledger_dir, file_name))
+                )
+
+        return sorted(chunks, key=lambda chunk: chunk[0])
+
+    def find_committed_ledger_chunk_covering(self, seqno):
+        for start_seqno, end_seqno, path in self.committed_ledger_chunks():
+            if end_seqno is not None and start_seqno <= seqno <= end_seqno:
+                return start_seqno, end_seqno, path
+        return None
+
+    def wait_for_committed_ledger_chunk_covering(self, seqno, timeout=5):
+        end_time = time.time() + timeout
+        while time.time() < end_time:
+            chunk = self.find_committed_ledger_chunk_covering(seqno)
+            if chunk is not None:
+                return chunk
+            time.sleep(0.1)
+
+        raise TimeoutError(
+            f"Could not find committed ledger chunk covering seqno {seqno} "
+            f"on node {self.local_node_id} after {timeout}s"
+        )
+
     def get_ledger_public_tables_at(self, seqno):
+        if not self.is_stopped():
+            LOG.warning(
+                "Parsing ledger files from live node {}.",
+                self.local_node_id,
+            )
         ledger = ccf.ledger.Ledger(self.remote.ledger_paths())
         assert ledger.last_committed_chunk_range[1] >= seqno
         tx = ledger.get_transaction(seqno)
         return tx.get_public_domain().get_tables()
 
     def get_ledger_public_state_at(self, seqno):
+        if not self.is_stopped():
+            LOG.warning(
+                "Parsing ledger files from live node {}.",
+                self.local_node_id,
+            )
         ledger = ccf.ledger.Ledger(self.remote.ledger_paths())
         assert ledger.last_committed_chunk_range[1] >= seqno
         return ledger.get_latest_public_state()
@@ -595,6 +643,20 @@ def get_ledger(self):
 
         return current_ledger_dir, [committed_ledger_dir]
 
+    def get_committed_ledger_dirs(self):
+        if not self.is_stopped():
+            LOG.warning(
+                "Copying committed ledger files from live node {}.",
+                self.local_node_id,
+            )
+        _, committed_ledger_dirs = self.get_ledger()
+        return committed_ledger_dirs
+
+    def get_committed_ledger(self, **kwargs):
+        kwargs.setdefault("committed_only", True)
+        kwargs.setdefault("contiguous_suffix", True)
+        return ccf.ledger.Ledger(self.get_committed_ledger_dirs(), **kwargs)
+
     def get_committed_snapshots(self, pre_condition_func=lambda src_dir, _: True):
         (
             main_snapshots_dir,

tests/jwt_test.py

@@ -8,7 +8,6 @@
 from contextlib import contextmanager
 
 import ca_certs
-import ccf.ledger
 import infra.clients
 import infra.crypto
 import infra.e2e_args
@@ -803,12 +802,10 @@ def check_refresh_progressed():
             timeout=max(5, args.jwt_key_refresh_interval_s * 5),
         )
 
-        # Force chunking
-        network.get_latest_ledger_public_state()
+        network.create_and_wait_for_chunk(primary)
         # Check that despite refreshing JWTs multiple times, only a single
         # transaction was created for this kid.
-        ledger_directories = primary.remote.ledger_paths()
-        ledger = ccf.ledger.Ledger(ledger_directories, contiguous_suffix=True)
+        ledger = primary.get_committed_ledger()
 
         last_key_refresh = None
         for chunk in ledger:

tests/lts_compatibility.py

@@ -4,7 +4,6 @@
 import json
 import os
 import shutil
-import time
 
 import ccf.ledger
 import infra.crypto
@@ -18,7 +17,6 @@
 import infra.proc
 import infra.utils
 import suite.test_requirements as reqs
-from ccf.tx_id import TxID
 from cryptography import x509
 from cryptography.hazmat.backends import default_backend
 from e2e_logging import test_random_receipts
@@ -192,24 +190,10 @@ def test_new_service(
     if test_jwt_cleanup:
 
         def get_fresh_public_state():
-            with primary.client() as c:
-                r = c.get("/node/commit")
-                target_seqno = TxID.from_str(r.body.json()["transaction_id"]).seqno
-            network.consortium.force_ledger_chunk(primary)
-            for _ in range(10):
-                ledger = ccf.ledger.Ledger(
-                    primary.remote.ledger_paths(),
-                    committed_only=True,
-                    contiguous_suffix=True,
-                )
-                public_state, last_seqno = ledger.get_latest_public_state()
-                if last_seqno >= target_seqno:
-                    return public_state
-
-                time.sleep(0.1)
-            assert (
-                False
-            ), f"Failed to up-to-date ledger state, seqno needed: {target_seqno}, last seqno: {last_seqno}"
+            # This still copies from a live node under the hood, but avoids the
+            # older pattern of parsing in-flight ledger chunks directly.
+            public_state, _ = network.get_latest_ledger_public_state()
+            return public_state
 
         def table_has_entries(table_name, public_state):
             rows = public_state.get(table_name, None)
@@ -548,7 +532,7 @@ def run_code_upgrade_from(
                 to_version,
                 expected_subject_name=service_subject_name,
             )
-            network.get_latest_ledger_public_state()
+            network.create_and_wait_for_chunk()
 
 
 @reqs.description("Run live compatibility with latest LTS")

tests/reconfiguration.py

@@ -886,9 +886,8 @@ def test_join_straddling_primary_replacement(network, args):
 def test_retired_nodes_stop_signing_after_retired_committed(network, args):
     primary, _ = network.find_primary()
 
-    # Force ledger flush of all transactions so far
-    network.get_latest_ledger_public_state()
-    ledger = ccf.ledger.Ledger(primary.remote.ledger_paths(), contiguous_suffix=True)
+    network.create_and_wait_for_chunk(primary)
+    ledger = primary.get_committed_ledger()
 
     # True once a subsequent signature has committed the retired_committed tx
     # itself: rc is then globally committed and the node's retired_committed
@@ -970,13 +969,13 @@ def test_add_node_with_read_only_ledger(network, args):
 
 @reqs.description("Confirm ledger contains expected entries")
 def test_ledger_invariants(network, args):
-    # Force ledger flush of all transactions so far
-    network.get_latest_ledger_public_state()
+    proposal, _ = network.create_and_wait_for_chunk()
 
     for node in network.nodes:
         LOG.info(f"Examining ledger on node {node.local_node_id}")
-        ledger_directories = node.remote.ledger_paths()
-        ledger = ccf.ledger.Ledger(ledger_directories, contiguous_suffix=True)
+        if node.is_joined():
+            node.wait_for_committed_ledger_chunk_covering(proposal.completed_seqno)
+        ledger = node.get_committed_ledger()
         check_signatures(ledger)
 
     return network
@@ -986,8 +985,7 @@ def test_ledger_invariants(network, args):
 def test_joining_nodes_snapshot_ledger_offset(network, args):
     primary, _ = network.find_primary()
 
-    network.consortium.force_ledger_chunk(primary)
-    network.get_latest_ledger_public_state()
+    network.create_and_wait_for_chunk(primary)
 
     network.txs.issue(network, number_txs=5, send_private=False, send_public=True)
     network.txs.issue(network, number_txs=5, send_private=False, send_public=True)
@@ -1029,14 +1027,10 @@ def test_joining_nodes_snapshot_ledger_offset(network, args):
         rest_txid,
     )
 
-    # flush ledger to disk from commit index
-    network.get_latest_ledger_public_state()
-    # Only use the committed ledger files flushed from above
-    _, committed_ledger_dirs = primary.get_ledger()
+    network.create_and_wait_for_chunk(primary, seqno=rest_txid.seqno)
+    committed_ledger_dirs = primary.get_committed_ledger_dirs()
     ledger = ccf.ledger.Ledger(
-        committed_ledger_dirs,
-        committed_only=True,
-        contiguous_suffix=True,
+        committed_ledger_dirs, committed_only=True, contiguous_suffix=True
     )
 
     snapshot_chunk_start = None

tests/recovery.py

@@ -2321,8 +2321,7 @@ def run_recover_snapshot_ledger_offset(args):
         network.start_and_open(args)
         primary, _ = network.find_primary()
 
-        network.consortium.force_ledger_chunk(primary)
-        network.get_latest_ledger_public_state()
+        network.create_and_wait_for_chunk(primary)
 
         network.txs.issue(network, number_txs=5, send_private=False, send_public=True)
         network.txs.issue(network, number_txs=5, send_private=False, send_public=True)
@@ -2360,7 +2359,7 @@ def run_recover_snapshot_ledger_offset(args):
         rest_txid = network.txs.issue(
             network, number_txs=5, send_private=False, send_public=True
         )
-        network.get_latest_ledger_public_state()
+        network.create_and_wait_for_chunk(primary, seqno=rest_txid.seqno)
 
         assert snapshot_trigger_txid.seqno < snapshot_seqno < rest_txid.seqno, (
             snapshot_trigger_txid,