feat: add meta and all-shards destination handling

multiversx_cross_shard_analysis/constants.py

@@ -1,12 +1,14 @@
 from enum import Enum
 
-
 origin_shard = "origin_shard"
 dest_shard = "dest_shard"
 meta = "meta"
 proposed = "proposed"
 committed = "committed"
 
+META_SHARD_ID = 4294967295
+ALL_SHARDS_ID = 4294967280
+
 MiniBlockTypes = Enum("MiniBlockType", [
     'MiniBlockHeaders',
     'ShardInfo',
@@ -47,7 +49,6 @@
     "meta_dest_exec_committed",
 ])
 
-
 # Mappings from field number to field name for MiniBlockHeaderReserved
 FIELD_NAME_MAPPING = {
     1: "ExecutionType",

multiversx_cross_shard_analysis/issues.py

@@ -1,6 +1,9 @@
 from enum import Enum
 from typing import Any, Callable
 
+from multiversx_cross_shard_analysis.constants import (ALL_SHARDS_ID,
+                                                       META_SHARD_ID)
+
 DEFAULT_MAX_ROUND_GAP_ALLOWED = 3
 DEFAULT_SUPERNOVA_ACTIVATION_EPOCH = 2
 
@@ -28,33 +31,74 @@ def check_missing_or_duplicate_destination(self, mb_info: dict[str, Any]) -> boo
         sender = mb_info.get("senderShardID")
         count = 0
 
-        for _, header in mb_info.get("mentioned", []):
+        for mtype, header in mb_info.get("mentioned", []):
+            # meta_origin mentions are origin notarizations, not destination registrations;
+            # for meta-destined miniblocks they also occur at shard_id == receiver
+            if mtype.startswith('meta_origin'):
+                continue
             if header.get("shard_id") == receiver and mb_info.get("type") in [0, 90]:
                 count += 1
 
         is_dest_missing = count == 0 and mb_info.get("type") in [0, 90]
-        is_dest_duplicate = count > 4 and mb_info.get("type") in [0, 90] and receiver != sender and mb_info.get("first_seen_epoch", 0) >= DEFAULT_SUPERNOVA_ACTIVATION_EPOCH
+        is_dest_duplicate = count > 4 and mb_info.get("type") in [0, 90] and receiver != sender and mb_info.get(
+            "first_seen_epoch", 0) >= DEFAULT_SUPERNOVA_ACTIVATION_EPOCH
 
         return is_dest_missing or is_dest_duplicate
 
     # Logic for: WRONG_PROCESSING_ORDER
     def check_wrong_order(self, mb_info: dict[str, Any]) -> bool:
+        if mb_info.get("receiverShardID") == ALL_SHARDS_ID:
+            return self.check_wrong_order_broadcast(mb_info)
+
         max_phase = -1
+        # for meta-destined miniblocks the dest proposal and the meta origin commit happen
+        # in the same metablock, so no ordering is asserted between them
+        is_meta_destined = mb_info.get("receiverShardID") == META_SHARD_ID
 
         for mtype, data in sorted(mb_info.get('mentioned', []), key=lambda x: x[1].get('round', 0)):
             if 'exec' in mtype:
                 phase = 1 if 'origin' in mtype else 4
             elif 'meta' in mtype:
                 phase = 2 if 'origin' in mtype else 5
             else:
-                phase = 0 if 'origin' in mtype else 3
+                phase = 0 if 'origin' in mtype else (2 if is_meta_destined else 3)
 
             if phase < max_phase:
                 return True
             max_phase = phase
 
         return False
 
+    # broadcast miniblocks reach every shard independently and meta notarizations of
+    # different shards interleave, so ordering is checked per destination shard;
+    # meta_dest mentions carry no destination shard id and are skipped
+    def check_wrong_order_broadcast(self, mb_info: dict[str, Any]) -> bool:
+        mentions = sorted(mb_info.get('mentioned', []), key=lambda x: x[1].get('round', 0))
+        dest_shards = {data.get('shard_id') for mtype, data in mentions if
+                       'origin' not in mtype and 'meta' not in mtype}
+
+        for shard in dest_shards:
+            max_phase = -1
+            for mtype, data in mentions:
+                if 'meta_dest' in mtype:
+                    continue
+                is_origin = 'origin' in mtype
+                if not is_origin and data.get('shard_id') != shard:
+                    continue
+
+                if 'exec' in mtype:
+                    phase = 1 if is_origin else 4
+                elif 'meta' in mtype:
+                    phase = 2
+                else:
+                    phase = 0 if is_origin else 3
+
+                if phase < max_phase:
+                    return True
+                max_phase = phase
+
+        return False
+
     def run_check(self, issue_type: 'Issues', mb_info: dict[str, Any]) -> bool:
         """Helper to route to the correct method."""
         check_map: dict[Issues, Callable] = {

multiversx_cross_shard_analysis/miniblock_data.py

@@ -1,7 +1,7 @@
-
 from typing import Any
 
-from multiversx_cross_shard_analysis.constants import (TYPE_NAMES, Colors)
+from multiversx_cross_shard_analysis.constants import (META_SHARD_ID,
+                                                       TYPE_NAMES, Colors)
 from multiversx_cross_shard_analysis.decode_reserved import \
     get_default_decoded_data
 from multiversx_cross_shard_analysis.issues import Issues
@@ -15,7 +15,8 @@ def __init__(self, miniblocks: dict[str, dict[str, Any]]):
 
     def verify_miniblocks(self) -> None:
         for mb_hash, mb_info in self.miniblocks.items():
-            mb_info['mentioned'] = sorted(mb_info.get('mentioned', []), key=lambda x: (x[1].get('epoch', 0), x[1].get('round', 0)))
+            mb_info['mentioned'] = sorted(mb_info.get('mentioned', []),
+                                          key=lambda x: (x[1].get('epoch', 0), x[1].get('round', 0)))
             mentioning_header = mb_info['mentioned'][0][1] if mb_info['mentioned'] else None
             if mentioning_header:
                 mb_info['first_seen_round'] = mentioning_header.get('round')
@@ -38,9 +39,11 @@ def get_color_for_state(self, mention_type: str, tx_count: int, header: dict[str
             reserved = get_default_decoded_data(tx_count=tx_count)
             if "meta" in mention_type:
                 if 'exec' in mention_type:
-                    color = Colors.meta_origin_exec_committed if mention_type.startswith('meta_origin') else Colors.meta_dest_exec_committed
+                    color = Colors.meta_origin_exec_committed if mention_type.startswith(
+                        'meta_origin') else Colors.meta_dest_exec_committed
                 else:
-                    color = Colors.meta_origin_committed if mention_type.startswith('meta_origin') else Colors.meta_dest_committed
+                    color = Colors.meta_origin_committed if mention_type.startswith(
+                        'meta_origin') else Colors.meta_dest_committed
             else:
                 if 'exec' in mention_type:
                     color = Colors.origin_exec_final if mention_type.startswith('origin') else Colors.dest_exec_final
@@ -51,16 +54,19 @@ def get_color_for_state(self, mention_type: str, tx_count: int, header: dict[str
             state = header.get('reserved', {}).get('State', '')
             if 'exec' in mention_type:
                 if state == 'Proposed':
-                    color = Colors.origin_exec_proposed if mention_type.startswith('origin') else Colors.dest_exec_proposed
+                    color = Colors.origin_exec_proposed if mention_type.startswith(
+                        'origin') else Colors.dest_exec_proposed
                 elif state == 'PartialExecuted':
-                    color = Colors.origin_exec_partial_executed if mention_type.startswith('origin') else Colors.dest_exec_partial_executed
+                    color = Colors.origin_exec_partial_executed if mention_type.startswith(
+                        'origin') else Colors.dest_exec_partial_executed
                 else:
                     color = Colors.origin_exec_final if mention_type.startswith('origin') else Colors.dest_exec_final
             else:
                 if state == 'Proposed':
                     color = Colors.origin_proposed if mention_type.startswith('origin') else Colors.dest_proposed
                 elif state == 'PartialExecuted':
-                    color = Colors.origin_partial_executed if mention_type.startswith('origin') else Colors.dest_partial_executed
+                    color = Colors.origin_partial_executed if mention_type.startswith(
+                        'origin') else Colors.dest_partial_executed
                 else:
                     color = Colors.origin_final if mention_type.startswith('origin') else Colors.dest_final
         return color
@@ -118,7 +124,9 @@ def get_data_for_detail_report(self) -> dict[str, list[dict[str, Any]]]:
                 reserved = header.get('reserved')
                 if reserved == {}:
                     reserved = get_default_decoded_data(tx_count=mb_info['txCount'])
-                mb_data['mentioned'][round_number].append((mention_type, f"txs {reserved['IndexOfFirstTxProcessed']}–{reserved['IndexOfLastTxProcessed']} / {mb_info['txCount']}", color))
+                mb_data['mentioned'][round_number].append((mention_type,
+                                                           f"txs {reserved['IndexOfFirstTxProcessed']}\u2013{reserved['IndexOfLastTxProcessed']} / {mb_info['txCount']}",
+                                                           color))
 
             if not origin_epoch:
                 print(f"Warning: origin_epoch not found for miniblock {mb_hash}")
@@ -178,7 +186,8 @@ def get_data_for_header_alarms_report(self) -> dict[int, Any]:
 
         seen_miniblocks = set[str]()
 
-        for mb_hash, mb_info in [(hash, miniblock) for hash, miniblock in self.miniblocks.items() if miniblock['hasAlarm']]:
+        for mb_hash, mb_info in [(hash, miniblock) for hash, miniblock in self.miniblocks.items() if
+                                 miniblock['hasAlarm']]:
             nonce = mb_info['nonce']
             shard_id = mb_info['senderShardID']
             epoch = mb_info['first_seen_epoch']
@@ -221,7 +230,9 @@ def get_data_for_header_alarms_report(self) -> dict[int, Any]:
                     report[epoch][issue][shard_id][nonce][round_number].append((label, mb_hash[:15] + '...', color))
                     seen_miniblocks.add(mb_hash)
 
-        for mb_hash in [item for item in self.miniblocks.keys() if item not in seen_miniblocks and self.miniblocks[item]['nonce'] in nonce_alarms.get(self.miniblocks[item]['senderShardID'], set())]:
+        for mb_hash in [item for item in self.miniblocks.keys() if
+                        item not in seen_miniblocks and self.miniblocks[item]['nonce'] in nonce_alarms.get(
+                                self.miniblocks[item]['senderShardID'], set())]:
             mb_info = self.miniblocks[mb_hash]
             nonce = mb_info['nonce']
             shard_id = mb_info['senderShardID']
@@ -315,9 +326,6 @@ def sort_report1(report: dict[int, dict[int, Any]]) -> dict[int, dict[int, Any]]
     return out
 
 
-META_SHARD_ID = 4294967295
-
-
 def sort_any(data: Any) -> Any:
     """
     Recursively sorts dictionaries and lists.

multiversx_cross_shard_analysis/test_issues.py

@@ -0,0 +1,142 @@
+from typing import Any
+
+from multiversx_cross_shard_analysis.constants import ALL_SHARDS_ID, META_SHARD_ID
+from multiversx_cross_shard_analysis.issues import Issues
+
+
+def mention(mtype: str, round_number: int, shard_id: int) -> tuple[str, dict[str, Any]]:
+    return (mtype, {"nonce": 0, "round": round_number, "epoch": 8, "shard_id": shard_id, "reserved": {}})
+
+
+def make_mb(receiver: int, mentioned: list[tuple[str, dict[str, Any]]], sender: int = 0, mb_type: int = 0) -> dict[
+    str, Any]:
+    return {
+        "receiverShardID": receiver,
+        "senderShardID": sender,
+        "type": mb_type,
+        "txCount": 1,
+        "first_seen_epoch": 8,
+        "mentioned": mentioned,
+    }
+
+
+# correct flow for a shard -> meta miniblock: the metablock proposes it for its own
+# execution (labeled dest_shard) and commits the origin exec result (meta_origin)
+# in the same round, then executes it in the next metablock
+def meta_destined_correct_flow(mb_type: int = 0) -> dict[str, Any]:
+    return make_mb(META_SHARD_ID, [
+        mention("origin_shard_proposed_headers_exec", 6379, 0),
+        mention("origin_shard_committed_headers_exec", 6379, 0),
+        mention("dest_shard_proposed_headers", 6380, META_SHARD_ID),
+        mention("meta_origin_shard_proposed_headers", 6380, META_SHARD_ID),
+        mention("dest_shard_committed_headers", 6380, META_SHARD_ID),
+        mention("meta_origin_shard_committed_headers", 6380, META_SHARD_ID),
+        mention("dest_shard_proposed_headers_exec", 6381, META_SHARD_ID),
+        mention("dest_shard_committed_headers_exec", 6381, META_SHARD_ID),
+    ], mb_type=mb_type)
+
+
+def normal_cross_shard_correct_flow() -> dict[str, Any]:
+    return make_mb(1, [
+        mention("origin_shard_proposed_headers_exec", 6379, 0),
+        mention("origin_shard_committed_headers_exec", 6379, 0),
+        mention("meta_origin_shard_proposed_headers", 6380, META_SHARD_ID),
+        mention("meta_origin_shard_committed_headers", 6380, META_SHARD_ID),
+        mention("dest_shard_proposed_headers", 6381, 1),
+        mention("dest_shard_committed_headers", 6381, 1),
+        mention("dest_shard_proposed_headers_exec", 6382, 1),
+        mention("dest_shard_committed_headers_exec", 6382, 1),
+        mention("meta_dest_shard_proposed_headers", 6383, META_SHARD_ID),
+        mention("meta_dest_shard_committed_headers", 6383, META_SHARD_ID),
+    ])
+
+
+class TestMissingOrDuplicateDestination:
+    issue = Issues.MISSING_OR_DUPLICATE_DESTINATION
+
+    def test_meta_destined_correct_flow_is_clean(self):
+        assert self.issue.check_missing_or_duplicate_destination(meta_destined_correct_flow()) is False
+
+    def test_meta_destined_scr_correct_flow_is_clean(self):
+        assert self.issue.check_missing_or_duplicate_destination(meta_destined_correct_flow(mb_type=90)) is False
+
+    def test_normal_cross_shard_correct_flow_is_clean(self):
+        assert self.issue.check_missing_or_duplicate_destination(normal_cross_shard_correct_flow()) is False
+
+    def test_missing_destination_detected(self):
+        mb = make_mb(1, [
+            mention("origin_shard_proposed_headers_exec", 6379, 0),
+            mention("origin_shard_committed_headers_exec", 6379, 0),
+            mention("meta_origin_shard_committed_headers", 6380, META_SHARD_ID),
+        ])
+        assert self.issue.check_missing_or_duplicate_destination(mb) is True
+
+    def test_duplicate_destination_detected(self):
+        mb = normal_cross_shard_correct_flow()
+        mb["mentioned"] += [
+            mention("dest_shard_proposed_headers", 6383, 1),
+            mention("dest_shard_committed_headers", 6383, 1),
+        ]
+        assert self.issue.check_missing_or_duplicate_destination(mb) is True
+
+    def test_meta_destined_double_include_detected(self):
+        mb = meta_destined_correct_flow()
+        mb["mentioned"] += [
+            mention("dest_shard_proposed_headers", 6382, META_SHARD_ID),
+            mention("dest_shard_committed_headers", 6382, META_SHARD_ID),
+        ]
+        assert self.issue.check_missing_or_duplicate_destination(mb) is True
+
+
+class TestWrongProcessingOrder:
+    issue = Issues.WRONG_PROCESSING_ORDER
+
+    def test_meta_destined_correct_flow_is_clean(self):
+        assert self.issue.check_wrong_order(meta_destined_correct_flow()) is False
+
+    def test_normal_cross_shard_correct_flow_is_clean(self):
+        assert self.issue.check_wrong_order(normal_cross_shard_correct_flow()) is False
+
+    def test_dest_proposal_before_meta_origin_commit_detected(self):
+        mb = make_mb(1, [
+            mention("origin_shard_committed_headers_exec", 6379, 0),
+            mention("dest_shard_proposed_headers", 6380, 1),
+            mention("meta_origin_shard_committed_headers", 6381, META_SHARD_ID),
+            mention("dest_shard_committed_headers_exec", 6382, 1),
+        ])
+        assert self.issue.check_wrong_order(mb) is True
+
+    def test_meta_destined_exec_before_proposal_detected(self):
+        mb = make_mb(META_SHARD_ID, [
+            mention("origin_shard_committed_headers_exec", 6379, 0),
+            mention("dest_shard_committed_headers_exec", 6380, META_SHARD_ID),
+            mention("dest_shard_committed_headers", 6381, META_SHARD_ID),
+            mention("meta_origin_shard_committed_headers", 6381, META_SHARD_ID),
+        ])
+        assert self.issue.check_wrong_order(mb) is True
+
+    # epoch-start peer (validator info) miniblock, broadcast from meta to all shards;
+    # shards include it at their own pace, so meta notarizations interleave with
+    # other shards' inclusions
+    def test_broadcast_interleaved_destinations_is_clean(self):
+        mb = make_mb(ALL_SHARDS_ID, [
+            mention("origin_shard_proposed_headers", 101, META_SHARD_ID),
+            mention("origin_shard_committed_headers", 101, META_SHARD_ID),
+            mention("dest_shard_proposed_headers", 102, 1),
+            mention("meta_dest_shard_proposed_headers", 103, META_SHARD_ID),
+            mention("meta_dest_shard_committed_headers", 103, META_SHARD_ID),
+            mention("dest_shard_committed_headers", 103, 1),
+            mention("dest_shard_proposed_headers", 103, 0),
+            mention("dest_shard_committed_headers", 103, 0),
+            mention("meta_dest_shard_proposed_headers", 104, META_SHARD_ID),
+            mention("meta_dest_shard_committed_headers", 104, META_SHARD_ID),
+        ], sender=META_SHARD_ID, mb_type=60)
+        assert self.issue.check_wrong_order(mb) is False
+
+    def test_broadcast_destination_before_origin_detected(self):
+        mb = make_mb(ALL_SHARDS_ID, [
+            mention("dest_shard_proposed_headers", 100, 1),
+            mention("origin_shard_committed_headers", 101, META_SHARD_ID),
+            mention("dest_shard_committed_headers", 102, 1),
+        ], sender=META_SHARD_ID, mb_type=60)
+        assert self.issue.check_wrong_order(mb) is True