feat(gossipsub-interop): add topic-streams HoL scenario

gossipsub-interop/Makefile

@@ -35,7 +35,7 @@ clean:
 	rm latest
 	rm plots/* || true
 
-test: test-partial-messages test-subnet-blob
+test: test-partial-messages test-subnet-blob test-topic-streams-hol
 
 test-partial-messages:
 	# Testing partial messages
@@ -95,6 +95,12 @@ test-subnet-blob:
 	@echo "Testing all"
 	uv run run.py --node_count 32 --composition go rust jvm nim && uv run checks/subnet_blob_msg.py latest/
 
+test-topic-streams-hol:
+	# Topic streams HoL: large then small on different topics; small must arrive first.
+	# Requires an implementation with topic streams (currently go).
+	@echo "Testing topic-streams HoL (go)"
+	@uv run run.py --node_count 2 --composition go --scenario "topic-streams-hol" && uv run checks/topic_streams_hol.py latest/
+
 test-go:
 	# Testing partial messages
 	@echo "Testing partial messages"
@@ -106,6 +112,9 @@ test-go:
 	@echo "Testing fanout"
 	@uv run run.py --node_count 2 --composition go --scenario "partial-messages-fanout" && uv run checks/partial_messages.py latest/
 
+	@echo "Testing topic-streams HoL"
+	@uv run run.py --node_count 2 --composition go --scenario "topic-streams-hol" && uv run checks/topic_streams_hol.py latest/
+
 
 test-rust-only:
 	# Testing partial messages
@@ -120,4 +129,4 @@ test-rust-only:
 
 
 
-.PHONY: binaries all clean test test-rust-only test-go test-subnet-blob test-partial-messages
+.PHONY: binaries all clean test test-rust-only test-go test-subnet-blob test-partial-messages test-topic-streams-hol

gossipsub-interop/README.md

@@ -86,6 +86,14 @@ uv run run.py --node_count 2 --composition go --scenario "partial-messages" && u
 
 That command runs the shadow simulation and then verifies the stdout logs have the expected message.
 
+Topic streams HoL (large then small on different topics; small must not wait for large):
+
+```bash
+uv run run.py --node_count 2 --composition go --scenario "topic-streams-hol" && uv run checks/topic_streams_hol.py latest/
+```
+
+This covers the [Topic Streams Extension](https://github.com/libp2p/specs/pull/729). Implementations without topic streams will HoL-block the small message behind the 1 MiB one; go (with topic streams) should not.
+
 ## Tests
 
 ```bash

gossipsub-interop/checks/topic_streams_hol.py

@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""Verify the small message was not HoL-blocked behind the large one.
+
+Scenario contract (topic-streams-hol):
+  - message id 1: 1 MiB on topic-large
+  - message id 2: 1 KiB on topic-small
+  - publisher is node 0; both topics are subscribed by all nodes
+
+With topic streams, receivers should log "Received Message" for the small
+message *before* the large one. Without topic streams, the small message is
+head-of-line blocked and arrives after the large message.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+from datetime import datetime
+from pathlib import Path
+
+LARGE_MESSAGE_ID = "1"
+SMALL_MESSAGE_ID = "2"
+PUBLISHER_NODE_ID = "0"
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description=(
+            "Validate that the small message arrived before the large message "
+            "on every non-publisher node (no cross-topic HoL blocking)."
+        )
+    )
+    parser.add_argument(
+        "shadow_output",
+        help="Path to the Shadow output directory (the one containing the hosts/ folder).",
+    )
+    parser.add_argument(
+        "--large-id",
+        default=LARGE_MESSAGE_ID,
+        help=f"Message id of the large message (default: {LARGE_MESSAGE_ID}).",
+    )
+    parser.add_argument(
+        "--small-id",
+        default=SMALL_MESSAGE_ID,
+        help=f"Message id of the small message (default: {SMALL_MESSAGE_ID}).",
+    )
+    parser.add_argument(
+        "--publisher",
+        default=PUBLISHER_NODE_ID,
+        help=f"Node id of the publisher to exclude (default: {PUBLISHER_NODE_ID}).",
+    )
+    return parser.parse_args()
+
+
+def parse_time(ts: str) -> datetime:
+    # Go slog RFC3339 / RFC3339Nano; Python 3.11+ handles offsets. Strip a
+    # trailing Z if present for fromisoformat compatibility on older Pythons.
+    if ts.endswith("Z"):
+        ts = ts[:-1] + "+00:00"
+    return datetime.fromisoformat(ts)
+
+
+def iter_stdout_logs(hosts_dir: Path):
+    for stdout_file in sorted(hosts_dir.rglob("*.stdout")):
+        if stdout_file.is_file():
+            yield stdout_file
+
+
+def first_receive_times(
+    hosts_dir: Path, large_id: str, small_id: str
+) -> dict[str, dict[str, datetime]]:
+    """Return node_id -> {message_id: first receive time} for large/small."""
+    wanted = {large_id, small_id}
+    # node_id -> msg_id -> datetime
+    times: dict[str, dict[str, datetime]] = {}
+
+    for log_path in iter_stdout_logs(hosts_dir):
+        node_name = log_path.parent.name  # e.g. "node0"
+        current_node_id: str | None = None
+
+        with log_path.open("r", encoding="utf-8", errors="replace") as fh:
+            for line in fh:
+                try:
+                    entry = json.loads(line)
+                except (json.JSONDecodeError, ValueError):
+                    continue
+
+                msg = entry.get("msg")
+                if msg == "PeerID":
+                    current_node_id = str(entry.get("node_id", node_name))
+                    times.setdefault(current_node_id, {})
+                elif msg == "Received Message":
+                    mid = str(entry.get("id", ""))
+                    if mid not in wanted:
+                        continue
+                    nid = current_node_id or node_name.removeprefix("node")
+                    ts_raw = entry.get("time")
+                    if not ts_raw:
+                        continue
+                    try:
+                        ts = parse_time(ts_raw)
+                    except ValueError:
+                        continue
+                    node_times = times.setdefault(nid, {})
+                    prev = node_times.get(mid)
+                    if prev is None or ts < prev:
+                        node_times[mid] = ts
+
+    return times
+
+
+def main() -> int:
+    args = parse_args()
+    base_dir = Path(args.shadow_output).expanduser().resolve()
+    if not base_dir.exists():
+        print(f"shadow output directory does not exist: {base_dir}", file=sys.stderr)
+        return 1
+
+    hosts_dir = base_dir / "hosts"
+    if not hosts_dir.is_dir():
+        print(f"hosts directory not found under: {base_dir}", file=sys.stderr)
+        return 1
+
+    times = first_receive_times(hosts_dir, args.large_id, args.small_id)
+    receivers = sorted(
+        nid for nid in times if nid != args.publisher
+    )
+
+    if not receivers:
+        print("no non-publisher nodes found in logs", file=sys.stderr)
+        return 1
+
+    print(
+        f"Checking HoL: small id={args.small_id} should arrive before "
+        f"large id={args.large_id} (publisher node {args.publisher})"
+    )
+    print(f"Receiver nodes: {len(receivers)}")
+    print()
+
+    failures: list[str] = []
+    for nid in receivers:
+        node_times = times[nid]
+        large_ts = node_times.get(args.large_id)
+        small_ts = node_times.get(args.small_id)
+
+        if large_ts is None and small_ts is None:
+            failures.append(nid)
+            print(f"  [FAIL] node{nid}: received neither message")
+            continue
+        if large_ts is None:
+            failures.append(nid)
+            print(f"  [FAIL] node{nid}: missing large message ({args.large_id})")
+            continue
+        if small_ts is None:
+            failures.append(nid)
+            print(f"  [FAIL] node{nid}: missing small message ({args.small_id})")
+            continue
+
+        delta = large_ts - small_ts
+        if small_ts < large_ts:
+            print(
+                f"  [OK]   node{nid}: small at {small_ts.isoformat()} "
+                f"before large at {large_ts.isoformat()} "
+                f"(small led by {delta.total_seconds():.3f}s)"
+            )
+        else:
+            failures.append(nid)
+            blocked_by = (small_ts - large_ts).total_seconds()
+            print(
+                f"  [FAIL] node{nid}: small at {small_ts.isoformat()} "
+                f"after large at {large_ts.isoformat()} "
+                f"(HoL blocked by {blocked_by:.3f}s)"
+            )
+
+    print()
+    if failures:
+        print(
+            f"FAILED: {len(failures)}/{len(receivers)} receivers saw HoL blocking "
+            f"or missing messages. Topic streams should deliver the small message "
+            f"before the large one.",
+            file=sys.stderr,
+        )
+        return 1
+
+    print(
+        f"PASSED: all {len(receivers)} receivers got the small message before "
+        f"the large message (no cross-topic HoL blocking)."
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

gossipsub-interop/experiment.py

@@ -246,6 +246,72 @@ def partial_message_fanout_scenario(
     return instructions
 
 
+def topic_streams_hol_scenario(
+    disable_gossip: bool, node_count: int
+) -> List[ScriptInstruction]:
+    """Publish a large message then a small one on different topics.
+
+    With topic streams, the small message should not be head-of-line blocked
+    behind the large one (receiver log timestamps: small before large).
+    Without topic streams, the small message arrives only after the large one.
+    """
+    if node_count < 2:
+        raise ValueError("topic-streams-hol requires at least 2 nodes")
+
+    instructions: List[ScriptInstruction] = []
+    gs_params = GossipSubParams()
+    if disable_gossip:
+        gs_params.Dlazy = 0
+        gs_params.GossipFactor = 0
+    instructions.extend(spread_heartbeat_delay(node_count, gs_params))
+
+    number_of_conns_per_node = min(20, node_count - 1)
+    instructions.extend(random_network_mesh(node_count, number_of_conns_per_node))
+
+    topic_large = "topic-large"
+    topic_small = "topic-small"
+    instructions.append(script_instruction.SubscribeToTopic(topicID=topic_large))
+    instructions.append(script_instruction.SubscribeToTopic(topicID=topic_small))
+
+    # Allow mesh formation / topic-stream negotiation before publishing.
+    elapsed_seconds = 30
+    instructions.append(script_instruction.WaitUntil(elapsedSeconds=elapsed_seconds))
+
+    publisher = 0
+    large_message_id = 1
+    small_message_id = 2
+    large_size = 1 * 1024 * 1024  # 1 MiB
+    small_size = 1 * 1024  # 1 KiB
+
+    # Back-to-back publishes: large first, then small. Implementations enqueue
+    # quickly so both are in flight concurrently on the wire when topic streams
+    # (or equivalent multiplexing) are available.
+    instructions.append(
+        script_instruction.IfNodeIDEquals(
+            nodeID=publisher,
+            instruction=script_instruction.Publish(
+                messageID=large_message_id,
+                topicID=topic_large,
+                messageSizeBytes=large_size,
+            ),
+        )
+    )
+    instructions.append(
+        script_instruction.IfNodeIDEquals(
+            nodeID=publisher,
+            instruction=script_instruction.Publish(
+                messageID=small_message_id,
+                topicID=topic_small,
+                messageSizeBytes=small_size,
+            ),
+        )
+    )
+
+    elapsed_seconds += 60
+    instructions.append(script_instruction.WaitUntil(elapsedSeconds=elapsed_seconds))
+    return instructions
+
+
 def scenario(
     scenario_name: str, node_count: int, disable_gossip: bool
 ) -> ExperimentParams:
@@ -334,6 +400,9 @@ def scenario(
                 )
             )
 
+        case "topic-streams-hol":
+            instructions = topic_streams_hol_scenario(disable_gossip, node_count)
+
         case _:
             raise ValueError(f"Unknown scenario name: {scenario_name}")