Assert ledger chunking (#7637)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Amaury Chamayou <amchamay@microsoft.com>

Author: 3 people

python/src/ccf/ledger.py

@@ -3,7 +3,7 @@
 
 import struct
 import os
-from enum import Enum
+from enum import Enum, Flag, auto
 
 from typing import NamedTuple, Optional, Tuple, Dict, List
 
@@ -79,6 +79,11 @@ def is_deprecated(self):
         )
 
 
+class TransactionFlags(Flag):
+    FORCE_CHUNK_AFTER = auto()
+    FORCE_CHUNK_BEFORE = auto()
+
+
 def to_uint_64(buffer):
     return struct.unpack("@Q", buffer)[0]
 

tests/e2e_operations.py

@@ -782,6 +782,7 @@ def test_ledger_chunk_redirect_recent(network, args):
         f"Dropping last ledger chunk {chunks[-1]} from late backup main ledger directory {main_ledger_dir}"
     )
     os.remove(os.path.join(main_ledger_dir, chunks[-1]))
+    network.skip_verify_chunking = True
     with late_backup.client(
         interface_name=infra.interfaces.FILE_SERVING_RPC_INTERFACE
     ) as c:
@@ -1464,6 +1465,7 @@ def run_empty_ledger_dir_check(args):
             # Now write a file in the directory
             with open(os.path.join(tmp_dir, "ledger_1000_1500.committed"), "wb") as f:
                 f.write(b"bar")
+            network.skip_verify_chunking = True
 
             # Start new network, this should fail
             try:

tests/governance.py

@@ -7,6 +7,7 @@
 import infra.path
 import infra.proc
 import infra.net
+from infra.node import CCFVersion
 import infra.e2e_args
 import infra.proposal
 import suite.test_requirements as reqs
@@ -508,7 +509,7 @@ def test_all_nodes_cert_renewal(network, args, valid_from=None):
     self_signed_node_certs_before = {}
     for node in network.get_joined_nodes():
         # Note: GET /node/self_signed_certificate endpoint was added after 2.0.0-r6
-        if node.version_after("ccf-2.0.0-rc6"):
+        if CCFVersion(node.version) > CCFVersion("ccf-2.0.0-rc6"):
             self_signed_node_certs_before[node.local_node_id] = (
                 node.retrieve_self_signed_cert()
             )
@@ -524,7 +525,7 @@ def test_all_nodes_cert_renewal(network, args, valid_from=None):
 
     for node in network.get_joined_nodes():
         node.set_certificate_validity_period(valid_from, validity_period_days)
-        if node.version_after("ccf-2.0.0-rc6"):
+        if CCFVersion(node.version) > CCFVersion("ccf-2.0.0-rc6"):
             assert (
                 self_signed_node_certs_before[node.local_node_id]
                 != node.retrieve_self_signed_cert()

tests/infra/consortium.py

@@ -9,6 +9,7 @@
 import infra.proc
 import infra.checker
 import infra.node
+from infra.node import CCFVersion
 import infra.crypto
 import infra.member
 from infra.proposal import ProposalState
@@ -436,7 +437,7 @@ def add_users_and_transition_service_to_open(self, remote_node, users):
             proposal["actions"].append({"name": "set_user", "args": {"cert": cert}})
 
         args = {}
-        if remote_node.version_after("ccf-2.0.0-rc3"):
+        if CCFVersion(remote_node.version) > CCFVersion("ccf-2.0.0-rc3"):
             args = {"args": {"next_service_identity": self.get_service_identity()}}
         proposal["actions"].append({"name": "transition_service_to_open", **args})
 
@@ -664,7 +665,7 @@ def transition_service_to_open(self, remote_node, previous_service_identity=None
                 is_recovery = False
 
         args = {}
-        if remote_node.version_after("ccf-2.0.0-rc3"):
+        if CCFVersion(remote_node.version) > CCFVersion("ccf-2.0.0-rc3"):
             args = {
                 "previous_service_identity": previous_service_identity,
                 "next_service_identity": self.get_service_identity(),
@@ -955,7 +956,7 @@ def check_for_service(self, remote_node, status, recovery_count=None):
             r = c.get("/node/network").body.json()
             current_status = r["service_status"]
             current_cert = r["service_certificate"]
-            if remote_node.version_after("ccf-2.0.3"):
+            if CCFVersion(remote_node.version) > CCFVersion("ccf-2.0.3"):
                 current_recovery_count = r["recovery_count"]
             else:
                 assert "recovery_count" not in r
@@ -974,7 +975,7 @@ def check_for_service(self, remote_node, status, recovery_count=None):
             assert (
                 current_status == status.value
             ), f"Service status {current_status} (expected {status.value})"
-            if remote_node.version_after("ccf-2.0.3"):
+            if CCFVersion(remote_node.version) > CCFVersion("ccf-2.0.3"):
                 assert (
                     recovery_count is None or current_recovery_count == recovery_count
                 ), f"Current recovery count {current_recovery_count} is not expected {recovery_count}"

tests/infra/jwt_issuer.py

@@ -13,6 +13,7 @@
 import uuid
 
 from infra.log_capture import flush_info
+from infra.node import CCFVersion
 from loguru import logger as LOG
 from enum import Enum
 from cryptography.x509 import load_pem_x509_certificate
@@ -308,7 +309,7 @@ def wait_for_refresh(self, network, args, kid=None):
         kid_ = kid or self.default_kid
         primary, _ = network.find_nodes()
         end_time = time.time() + timeout
-        if primary.version_after("ccf-5.0.0-rc3"):
+        if CCFVersion(primary.version) > CCFVersion("ccf-5.0.0-rc3"):
             with primary.api_versioned_client(
                 network.consortium.get_any_active_member().local_id,
                 api_version=args.gov_api_version,
@@ -343,7 +344,7 @@ def wait_for_refresh(self, network, args, kid=None):
                     keys = r.body.json()
                     if kid_ in keys:
                         kid_vals = keys[kid_]
-                        if primary.version_after("ccf-5.0.0-dev17"):
+                        if CCFVersion(primary.version) > CCFVersion("ccf-5.0.0-dev17"):
                             assert len(kid_vals) == 1
                             stored_cert = kid_vals[0]["cert"]
                         else:

tests/infra/member.py

@@ -6,6 +6,7 @@
 import infra.proposal
 import infra.crypto
 import infra.clients
+from infra.node import CCFVersion
 import http
 import os
 import base64
@@ -152,7 +153,7 @@ def __init__(self):
 
         def _by_node_version(self, remote_node):
             min_version = "4.0.0"
-            if remote_node.version_after(min_version):
+            if CCFVersion(remote_node.version) > CCFVersion(min_version):
                 return self._preview_v1
             else:
                 raise ValueError(

tests/infra/network.py

@@ -11,6 +11,7 @@
 import infra.proc
 import infra.service_load
 import infra.node
+from infra.node import CCFVersion
 import infra.consortium
 import infra.e2e_args
 import ccf.ledger
@@ -23,6 +24,8 @@
 import functools
 import re
 import hashlib
+import json
+
 from datetime import datetime, timedelta, timezone
 from infra.consortium import slurp_file
 from collections import deque
@@ -233,6 +236,7 @@ def __init__(
         service_load=None,
         node_data_json_file=None,
         next_node_id=0,
+        skip_verify_chunking=False,
     ):
         # Map of node id to dict of node arg to override value
         # for example, to set the election timeout to 2s for node 3:
@@ -249,6 +253,7 @@ def __init__(
             self.service_load = service_load
             self.recovery_count = 0
             self.common_dir = None
+            self.skip_verify_chunking = skip_verify_chunking
         else:
             self.consortium = existing_network.consortium
             self.users = existing_network.users
@@ -264,6 +269,7 @@ def __init__(
                 self.service_load.set_network(self)
             self.recovery_count = existing_network.recovery_count
             self.common_dir = existing_network.common_dir
+            self.skip_verify_chunking = existing_network.skip_verify_chunking
 
         self.ignoring_shutdown_errors = False
         self.ignore_error_patterns = []
@@ -383,7 +389,10 @@ def _setup_node(
             current_ledger_dir, committed_ledger_dirs = target_node.get_ledger()
 
         # Note: temporary fix until second snapshot directory is ported to 2.x branch
-        if not node.version_after("ccf-2.0.3") and read_only_snapshots_dir is not None:
+        if (
+            not CCFVersion(node.version) > CCFVersion("ccf-2.0.3")
+            and read_only_snapshots_dir is not None
+        ):
             snapshots_dir = read_only_snapshots_dir
 
         node.prepare_join(
@@ -940,7 +949,7 @@ def recover(
         self.consortium.activate(random_node)
         expected_status = (
             ServiceStatus.RECOVERING
-            if random_node.version_after("ccf-2.0.0-rc3")
+            if CCFVersion(random_node.version) > CCFVersion("ccf-2.0.0-rc3")
             else ServiceStatus.OPENING
         )
         self.consortium.check_for_service(
@@ -1043,27 +1052,80 @@ def list_files_in_dirs_with_checksums(dirs):
             if last_ledger_seqno > longest_ledger_seqno:
                 assert longest_ledger_files is None or longest_ledger_files.issubset(
                     ledger_files
-                ), f"Ledger files on node {longest_ledger_node.local_node_id} do not match files on node {node.local_node_id}: {longest_ledger_files}, expected subset of {ledger_files}, diff: {ledger_files - longest_ledger_files}"
+                ), f"Ledger files on node {longest_ledger_node.local_node_id} do not match files on node {node.local_node_id}: {longest_ledger_files}, expected subset of {ledger_files}, diff: (Only on {node.local_node_id}: {ledger_files - longest_ledger_files}, Only on {longest_ledger_node.local_node_id}: {longest_ledger_files - ledger_files})"
                 longest_ledger_files = ledger_files
                 longest_ledger_node = node
                 longest_ledger_seqno = last_ledger_seqno
             else:
                 assert ledger_files.issubset(
                     longest_ledger_files
-                ), f"Ledger files on node {node.local_node_id} do not match files on node {longest_ledger_node.local_node_id}: {ledger_files}, expected subset of {longest_ledger_files}, diff: {longest_ledger_files - ledger_files}"
+                ), f"Ledger files on node {node.local_node_id} do not match files on node {longest_ledger_node.local_node_id}: {ledger_files}, expected subset of {longest_ledger_files}, diff: (Only on {longest_ledger_node.local_node_id}: {longest_ledger_files - ledger_files}, Only on {node.local_node_id}: {ledger_files - longest_ledger_files})"
 
         if longest_ledger_files:
             LOG.info(
                 f"Verified {len(longest_ledger_files)} ledger files consistency on all {len(self.nodes)} stopped nodes"
             )
 
+    def check_ledger_files_chunk_flags(self):
+        for node in self.nodes:
+            if node.remote is None:
+                continue
+            ledger_paths = node.remote.ledger_paths()
+            for path in ledger_paths:
+                ledger = ccf.ledger.Ledger([path])
+                chunks = list(ledger)
+                for cur, nxt in zip([None] + chunks, chunks + [None]):
+                    if cur is None:
+                        continue
+
+                    if nxt is None:
+                        # Assume that the next chunk would emit chunk_before
+                        flag_force_chunk_before = True
+
+                    else:
+                        nxt_tx = nxt[0]
+                        flags = ccf.ledger.TransactionFlags(
+                            nxt_tx.get_transaction_header().flags
+                        )
+                        flag_force_chunk_before = (
+                            ccf.ledger.TransactionFlags.FORCE_CHUNK_BEFORE in flags
+                        )
+                        if flag_force_chunk_before:
+                            # We should only ever emit force_chunk_before if this is the genesis transaction of a recovering service
+                            # Otherwise this tx could be rolled back, breaking the consistency of chunking across the network
+                            tables = nxt_tx.get_public_domain().get_tables()
+                            assert "public:ccf.gov.service.info" in tables
+                            service_info = json.loads(
+                                tables["public:ccf.gov.service.info"][
+                                    b"\x00\x00\x00\x00\x00\x00\x00\x00"
+                                ]
+                            )
+                            assert (
+                                "status" in service_info
+                                and service_info["status"] == "Recovering"
+                            ), f"Node {node.local_node_id} has a chunk which forces chunking before but does not recover the service: {nxt.filename()}"
+
+                    last_tx = cur[-1]
+                    flags = ccf.ledger.TransactionFlags(
+                        last_tx.get_transaction_header().flags
+                    )
+                    flag_force_chunk_after = (
+                        ccf.ledger.TransactionFlags.FORCE_CHUNK_AFTER in flags
+                    )
+
+                    assert (
+                        flag_force_chunk_after or flag_force_chunk_before
+                    ), f"Node {node.local_node_id} has chunks which do not force chunking correctly: {cur.filename()} -> {nxt.filename() if nxt is not None else 'NA'}"
+
     def stop_all_nodes(
         self,
         skip_verification=False,
         verbose_verification=False,
         accept_ledger_diff=False,
+        skip_verify_chunking=None,
         **kwargs,
     ):
+        skip_verify_chunking = skip_verify_chunking or self.skip_verify_chunking
         if not skip_verification and self.txs is not None:
             LOG.info("Verifying that all committed txs can be read before shutdown")
             log_capture = []
@@ -1102,6 +1164,10 @@ def stop_all_nodes(
         if not accept_ledger_diff:
             self.check_ledger_files_identical(**kwargs)
 
+        if not skip_verify_chunking:
+            LOG.info("Verifying ledger chunk flags before shutdown")
+            self.check_ledger_files_chunk_flags()
+
         if fatal_error_found:
             if self.ignoring_shutdown_errors:
                 LOG.warning("Ignoring shutdown errors")
@@ -1254,7 +1320,7 @@ def retire_node(self, remote_node, node_to_retire, timeout=10):
         )
         if remote_node == node_to_retire:
             remote_node, _ = self.wait_for_new_primary(remote_node)
-        if remote_node.version_after("ccf-2.0.4") and not pending:
+        if CCFVersion(remote_node.version) > CCFVersion("ccf-2.0.4") and not pending:
             end_time = time.time() + timeout
             r = None
             while time.time() < end_time:
@@ -1996,7 +2062,9 @@ def close_on_error(net, pdb=False):
             pdb.post_mortem()
 
         LOG.info("Stopping network")
-        net.stop_all_nodes(skip_verification=True, accept_ledger_diff=True)
+        net.stop_all_nodes(
+            skip_verification=True, accept_ledger_diff=True, skip_verify_chunking=True
+        )
 
         raise
 
@@ -2014,6 +2082,7 @@ def network(
     version=None,
     service_load=None,
     node_data_json_file=None,
+    skip_verify_chunking=False,
     **kwargs,
 ):
     """
@@ -2042,11 +2111,15 @@ def network(
         version=version,
         service_load=service_load,
         node_data_json_file=node_data_json_file,
+        skip_verify_chunking=skip_verify_chunking,
         **kwargs,
     )
     with close_on_error(net, pdb=pdb):
         yield net
     LOG.info("Stopping network")
-    net.stop_all_nodes(skip_verification=True, accept_ledger_diff=True)
+    net.stop_all_nodes(
+        skip_verification=True,
+        accept_ledger_diff=True,
+    )
     if init_partitioner:
         net.partitioner.cleanup()

tests/infra/node.py

@@ -3,6 +3,7 @@
 
 from contextlib import contextmanager, closing
 from enum import Enum, auto
+import functools
 import infra.crypto
 import infra.remote
 from datetime import datetime, timedelta, timezone
@@ -107,6 +108,28 @@ def version_after(version, cmp_version):
     ) > ccf._versionifier.to_python_version(cmp_version)
 
 
+@functools.total_ordering
+class CCFVersion:
+    # None is assumed to be the latest development version
+    # so None > any specific version, and None == None
+    def __init__(self, version_str):
+        self.version_str = version_str
+        if version_str is not None:
+            self.parsed_version = ccf._versionifier.to_python_version(version_str)
+        else:
+            self.parsed_version = None
+
+    def __eq__(self, other):
+        return self.parsed_version == other.parsed_version
+
+    def __lt__(self, other):
+        if self.parsed_version is None:
+            return False
+        if other.parsed_version is None:
+            return True
+        return self.parsed_version < other.parsed_version
+
+
 class Node:
     def __init__(
         self,
@@ -823,7 +846,7 @@ def check_log_for_error_message(self, msg):
         return False
 
     def version_after(self, version):
-        return version_after(self.version, version)
+        return CCFVersion(self.version) > CCFVersion(version)
 
     def get_receipt(self, view, seqno, timeout=3):
         found = False