sim.rs_mesh: heap-backed adjacency and stats, 16-node stress ring (#21)

Author: ch4r10t33r

README.md

@@ -27,29 +27,29 @@ Zig helpers for the wire formats of **[ethp2p](https://github.com/ethp2p/ethp2p)
 | RS emit planner (fair dispatch heap) | [`broadcast/rs/emit.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/rs/emit.go) | `layer.emit_planner` |
 | RS parity encode (klauspost-default matrix) | [klauspost/reedsolomon](https://github.com/klauspost/reedsolomon) via ethp2p RS strategy | `layer.rs_encode`, `ReedSolomon`, `decodeMessage` |
 | RS unified strategy (per-session) | [`broadcast/rs/strategy.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/rs/strategy.go) | `layer.rs_strategy` |
-| Abstract RS mesh (strategy-only; Go `TestNetwork` 0–1, 4-node ring, eight-node ring under stress) | [`sim/scenario_test.go`](https://github.com/ethp2p/ethp2p/blob/main/sim/scenario_test.go) | `sim.rs_mesh`, `zig build simtest` |
+| Abstract RS mesh (heap-backed adjacency + `PeerSessionStats`; cap `MaxMeshNodes`; Go `TestNetwork` 0–1, 4-node ring; 8- / 16-node rings under stress) | [`sim/scenario_test.go`](https://github.com/ethp2p/ethp2p/blob/main/sim/scenario_test.go) | `sim.rs_mesh`, `zig build simtest` |
 | Gossipsub sim publish bytes (same layout as Go `encodeGossipsubMessage`) + default topic | [`sim/strategy_gossipsub.go`](https://github.com/ethp2p/ethp2p/blob/main/sim/strategy_gossipsub.go) | `sim.gossipsub_transport` |
 | Abstract topic fanout + per-peer inboxes (no protobuf RPC) | same driver `Publish` / subscribe mesh | `sim.gossipsub_protocol` |
 | Encode + fanout helper | `GossipsubNode.Publish` + topic delivery | `sim.gossipsub_broadcast` |
 | App payload envelope entry point (re-export) | `encodeGossipsubMessage` | `broadcast.gossip` |
 | Gossipsim cross-checks (golden envelope, mesh fanout, `broadcast.gossip` vs transport) | — | `sim.gossipsub_interop` |
 | Gossipsub `ControlIHave` / `ControlIWant` protobuf bodies (subset of [libp2p `rpc.proto`](https://github.com/libp2p/go-libp2p-pubsub/blob/master/pb/rpc.proto)) | `ControlMessage` nested fields | `sim.gossipsub_rpc_pb`, `proto/gossipsub_rpc.proto` |
 | Gossipsub top-level `RPC` with `control` only (field 3) | length-delimited `RPC` shell for stream payloads | `sim.gossipsub_rpc_pb` (`encodeRpcEnvelopeControl`, `decodeRpcControlOnly`) |
-| **Still open** (see [issues](#pending-work)) | Full gossipsub `RPC`, libp2p/simnet host, RLNC, larger RS mesh graphs, optional channel-style event loop / `VerdictPending` for non-RS schemes | — |
+| **Still open** (see [issues](#pending-work)) | Full gossipsub `RPC`, libp2p/simnet host, RLNC, optional channel-style event loop / `VerdictPending` for non-RS schemes | — |
 
 ## Pending work
 
-**On `main` today:** wire + layer RS strategy; `layer.dedup` / `layer.dedup_registry` / `layer.verify_queue` / `layer.verify_workers`; `broadcast.*` (engine, channel, `relay_async_verify`, verified + unverified relay ingest); abstract RS mesh (2-, 4-, 6-node; optional **stress** adds six-node budget and **eight-node ring**); gossipsim stack; gossipsub `ControlIHave` / `ControlIWant` plus **`RPC` control-only envelope** helpers in `sim.gossipsub_rpc_pb`. CI enforces `build.zig.zon` `minimum_zig_version` vs workflow `ZIG_VERSION`; `just check-zig-ci-align` matches locally. Default `zig build test` stays fast.
+**On `main` today:** wire + layer RS strategy; `layer.dedup` / `layer.dedup_registry` / `layer.verify_queue` / `layer.verify_workers`; `broadcast.*` (engine, channel, `relay_async_verify`, verified + unverified relay ingest); abstract RS mesh (**heap-backed**, 2-, 4-, 6-node default; **stress** adds higher six-node budget plus **8- and 16-node rings**); gossipsim stack; gossipsub `ControlIHave` / `ControlIWant` plus **`RPC` control-only envelope** helpers in `sim.gossipsub_rpc_pb`. CI enforces `build.zig.zon` `minimum_zig_version` vs workflow `ZIG_VERSION`; `just check-zig-ci-align` matches locally. Default `zig build test` stays fast.
 
-**Suggested next:** [#13](https://github.com/ch4r10t33r/zig-ethp2p/issues/13) — extend `sim/rs_mesh.zig` with larger topologies and budgets aligned with Go `sim/scenario_test.go` (builds on existing mesh tests without new transport).
+**Suggested next:** [#12](https://github.com/ch4r10t33r/zig-ethp2p/issues/12) — full gossipsub `RPC`, libp2p streams, simnet/QUIC-style host.
 
 **Tracked issues** (roadmap, not exhaustive):
 
 | Issue | Status / topic |
 |-------|----------------|
 | [#11](https://github.com/ch4r10t33r/zig-ethp2p/issues/11) | **Closed** — engine dedup + async verify ingest (`relayIngestChunk*Engine`, `RelayAsyncVerifier.initBound`, `sessionDecodeClearEngineDedup`); RS verify stays sync (no `VerdictPending`) |
 | [#12](https://github.com/ch4r10t33r/zig-ethp2p/issues/12) | Open — full gossipsub `RPC`, libp2p streams, simnet/QUIC-style host |
-| [#13](https://github.com/ch4r10t33r/zig-ethp2p/issues/13) | Open — RS abstract mesh: larger graphs & Go scalability (**in progress**) |
+| [#13](https://github.com/ch4r10t33r/zig-ethp2p/issues/13) | **Closed** — heap-backed `sim/rs_mesh` (`MaxMeshNodes`), 16-node ring stress (Go `TestNetwork` remains the default topology reference) |
 | [#14](https://github.com/ch4r10t33r/zig-ethp2p/issues/14) | Open — RLNC and additional EC `Scheme` types |
 | [#15](https://github.com/ch4r10t33r/zig-ethp2p/issues/15) | **Closed** — `UPSTREAM.md` Zig toolchain note, CI `minimum_zig_version` check, `just check-zig-ci-align` |
 
@@ -63,7 +63,7 @@ Zig helpers for the wire formats of **[ethp2p](https://github.com/ethp2p/ethp2p)
 zig build
 zig build test         # wire, layer, broadcast, sim (default CI)
 zig build simtest      # alias of `test` (mesh-focused name)
-zig build test-stress  # same tests with `ZIG_ETHP2P_STRESS=1` (longer RS mesh case)
+zig build test-stress  # `ZIG_ETHP2P_STRESS=1` (longer RS mesh + 8-/16-node ring cases)
 ```
 
 Add as a dependency and import the module `zig_ethp2p` (see `build.zig`).

UPSTREAM.md

@@ -26,7 +26,7 @@ When updating:
 
 ## Abstract mesh tests
 
-`src/sim/rs_mesh.zig` runs the same RS **settings and graph topologies** as `sim/scenario_test.go` (`TestNetwork` RS / RS-ChunkLen) against `layer.RsStrategy` in-process (no libp2p, no Go simnet). `zig build test` and `zig build simtest` both execute them. With `ZIG_ETHP2P_STRESS=1` (see `zig build test-stress`), an additional 6-node case uses a higher round budget.
+`src/sim/rs_mesh.zig` runs the same RS **settings and graph topologies** as `sim/scenario_test.go` (`TestNetwork` RS / RS-ChunkLen) against `layer.RsStrategy` in-process (no libp2p, no Go simnet). Adjacency and per-peer stats are **heap-allocated** (`MaxMeshNodes` cap). `zig build test` and `zig build simtest` both execute them. With `ZIG_ETHP2P_STRESS=1` (see `zig build test-stress`), extra cases use higher round budgets and add **8- and 16-node** ring graphs beyond the Go file’s largest fixed topology.
 
 ## Specifications
 

build.zig

@@ -33,6 +33,6 @@ pub fn build(b: *std.Build) void {
     const run_stress = b.addRunArtifact(lib_tests);
     run_stress.setEnvironmentVariable("ZIG_ETHP2P_STRESS", "1");
     run_stress.has_side_effects = true;
-    const stress_step = b.step("test-stress", "Run tests with ZIG_ETHP2P_STRESS=1 (longer RS mesh)");
+    const stress_step = b.step("test-stress", "Run tests with ZIG_ETHP2P_STRESS=1 (longer RS mesh, 8-/16-node rings)");
     stress_step.dependOn(&run_stress.step);
 }

src/sim/rs_mesh.zig

@@ -1,6 +1,7 @@
 //! Abstract RS mesh (“simnet-style”) exercises `layer.RsStrategy` with the same RS settings and
 //! topologies as ethp2p `sim/scenario_test.go` `TestNetwork`, without libp2p or Go simnet.
 //!
+//! Graph structure and `PeerSessionStats` matrices are heap-backed (`MaxMeshNodes` upper bound).
 //! This validates zig-ethp2p’s strategy layer under multi-hop chunk forwarding; it is not a UDP
 //! timing simulation.
 
@@ -40,26 +41,50 @@ const MeshNode = struct {
     strat: RsStrategy = undefined,
 };
 
+/// Upper bound on `MeshParams.node_count` for abstract mesh sims (heap use is O(n²) for stats + edges).
+pub const MaxMeshNodes = 256;
+
 /// Run abstract RS broadcast until every non-origin node decodes `params.payload`, or `max_rounds`.
 pub fn runAbstractRsMesh(gpa: Allocator, params: MeshParams) !void {
     const n = params.node_count;
     std.debug.assert(n >= 2);
-    std.debug.assert(n <= MaxNodes);
+    if (n > MaxMeshNodes) return error.MeshTooLarge;
 
     var nodes = try gpa.alloc(MeshNode, n);
     defer gpa.free(nodes);
 
-    var deg: [MaxNodes]u8 = .{0} ** MaxNodes;
-    var neigh: [MaxNodes][MaxNodes]u8 = .{.{0} ** MaxNodes} ** MaxNodes;
+    var deg = try gpa.alloc(usize, n);
+    defer gpa.free(deg);
+    @memset(deg, 0);
 
     for (params.edges) |e| {
         std.debug.assert(e.a < n and e.b < n and e.a != e.b);
-        neigh[e.a][deg[e.a]] = @intCast(e.b);
         deg[e.a] += 1;
-        neigh[e.b][deg[e.b]] = @intCast(e.a);
         deg[e.b] += 1;
     }
 
+    var neigh = try gpa.alloc([]usize, n);
+    defer {
+        for (neigh) |row| gpa.free(row);
+        gpa.free(neigh);
+    }
+    for (0..n) |i| {
+        neigh[i] = try gpa.alloc(usize, deg[i]);
+    }
+
+    var cursor = try gpa.alloc(usize, n);
+    defer gpa.free(cursor);
+    @memset(cursor, 0);
+    @memset(deg, 0);
+
+    for (params.edges) |e| {
+        neigh[e.a][cursor[e.a]] = e.b;
+        cursor[e.a] += 1;
+        neigh[e.b][cursor[e.b]] = e.a;
+        cursor[e.b] += 1;
+    }
+    for (0..n) |i| deg[i] = neigh[i].len;
+
     for (nodes, 0..) |*node, i| {
         node.id = nodeId(&node.id_buf, i);
     }
@@ -79,13 +104,18 @@ pub fn runAbstractRsMesh(gpa: Allocator, params: MeshParams) !void {
         nodes[built].strat = try RsStrategy.newRelay(gpa, params.cfg, &nodes[0].strat.preamble);
     }
 
-    var stats: [MaxNodes][MaxNodes]broadcast_types.PeerSessionStats = undefined;
-    for (&stats) |*row| {
-        for (row) |*cell| cell.* = .{ .peer_id = &.{} };
+    var stats = try gpa.alloc([]broadcast_types.PeerSessionStats, n);
+    defer {
+        for (stats) |row| gpa.free(row);
+        gpa.free(stats);
+    }
+    for (0..n) |i| {
+        stats[i] = try gpa.alloc(broadcast_types.PeerSessionStats, n);
+        for (stats[i]) |*cell| cell.* = .{ .peer_id = &.{} };
     }
 
     for (nodes, 0..) |*dst, di| {
-        var k: u8 = 0;
+        var k: usize = 0;
         while (k < deg[di]) : (k += 1) {
             const j = neigh[di][k];
             try dst.strat.attachPeer(nodes[j].id, &stats[di][j]);
@@ -114,7 +144,7 @@ pub fn runAbstractRsMesh(gpa: Allocator, params: MeshParams) !void {
             }
         }
 
-        try exchangeRouting(gpa, nodes[0..n], neigh, deg);
+        try exchangeRouting(gpa, nodes[0..n], neigh[0..n], deg[0..n]);
 
         var all = true;
         for (1..n) |j| {
@@ -137,20 +167,19 @@ pub fn runAbstractRsMesh(gpa: Allocator, params: MeshParams) !void {
     }
 }
 
-const MaxNodes = 8;
-
 fn exchangeRouting(
     gpa: Allocator,
     nodes: []MeshNode,
-    neigh: [MaxNodes][MaxNodes]u8,
-    deg: [MaxNodes]u8,
+    neigh: []const []const usize,
+    deg: []const usize,
 ) !void {
+    std.debug.assert(neigh.len == nodes.len and deg.len == nodes.len);
     for (nodes, 0..) |*src, si| {
         const pr = try src.strat.pollRouting(gpa, false);
         if (!pr.emit) continue;
         const bm = pr.bitmap orelse continue;
         defer bm.deinit(gpa);
-        var k: u8 = 0;
+        var k: usize = 0;
         while (k < deg[si]) : (k += 1) {
             const j = neigh[si][k];
             const cancels = try nodes[j].strat.routingUpdate(src.id, bm);
@@ -322,3 +351,34 @@ test "abstract RS mesh eight nodes ring env stress (large-network scale)" {
         .max_rounds = 8_000_000,
     });
 }
+
+fn ringEdges(comptime node_count: usize) [node_count]Edge {
+    var out: [node_count]Edge = undefined;
+    for (0..node_count) |i| {
+        out[i] = .{ .a = i, .b = (i + 1) % node_count };
+    }
+    return out;
+}
+
+// Heap-backed mesh supports rings beyond the old fixed `MaxNodes`; stress-only.
+test "abstract RS mesh sixteen nodes ring env stress" {
+    const builtin = @import("builtin");
+    if (builtin.os.tag == .windows) return;
+
+    const gpa = std.testing.allocator;
+    const env = std.process.getEnvVarOwned(gpa, "ZIG_ETHP2P_STRESS") catch return;
+    defer gpa.free(env);
+    if (!std.mem.eql(u8, env, "1")) return;
+
+    const edges = comptime ringEdges(16);
+    var payload: [10 * 1024]u8 = undefined;
+    fillPayload(&payload);
+
+    try runAbstractRsMesh(gpa, .{
+        .node_count = 16,
+        .edges = &edges,
+        .cfg = rsCfgDefault(),
+        .payload = &payload,
+        .max_rounds = 25_000_000,
+    });
+}