sync: stop wedging on a false "peer does not support blocks_by_range"

Dockerfile

@@ -95,13 +95,17 @@ COPY .git/refs .git/refs
 #   amd64: x86-64-v3 (AVX2, no AVX512) — portable across modern x86_64 servers
 #   arm64: generic CPU — QEMU-compatible when cross-building
 ARG TARGETARCH
+# Set to true to compile in the experimental ethp2p parallel RS-broadcast
+# transport (`docker build --build-arg ETHP2P=true ...`). Default off: the
+# adapter is comptime-excluded and its `zig_ethp2p` dependency is not fetched.
+ARG ETHP2P=false
 RUN --mount=type=cache,target=/root/.cache/zig \
     --mount=type=cache,target=/root/.cargo/registry \
     --mount=type=cache,target=/root/.cargo/git \
     --mount=type=cache,target=/app/rust/target \
-    EXTRA_ZIG_FLAGS="" && \
+    EXTRA_ZIG_FLAGS="-Dethp2p=${ETHP2P}" && \
     if [ "$TARGETARCH" = "amd64" ]; then \
-        EXTRA_ZIG_FLAGS="-Dcpu=x86_64_v3 -Drust-target-cpu=x86-64-v3"; \
+        EXTRA_ZIG_FLAGS="$EXTRA_ZIG_FLAGS -Dcpu=x86_64_v3 -Drust-target-cpu=x86-64-v3"; \
     elif [ "$TARGETARCH" = "arm64" ]; then \
         export RUSTFLAGS="-C target-cpu=generic"; \
     fi && \

build.zig

@@ -167,6 +167,11 @@ pub fn build(b: *Builder) !void {
     build_options.addOption([]const u8, "prover", @tagName(prover));
     build_options.addOption(bool, "has_risc0", prover == .risc0 or prover == .all);
     build_options.addOption(bool, "has_openvm", prover == .openvm or prover == .all);
+    // Optional parallel ethp2p RS-broadcast transport (off by default). When
+    // false the adapter selector picks a stub and `zig_ethp2p` is never
+    // imported, so the default binary is unchanged.
+    const ethp2p_enabled = b.option(bool, "ethp2p", "Compile in the experimental ethp2p parallel RS-broadcast transport (default: false)") orelse false;
+    build_options.addOption(bool, "ethp2p", ethp2p_enabled);
     // Absolute path to test-keys for pre-generated validator keys
     build_options.addOption([]const u8, "test_keys_path", b.pathFromRoot("test-keys/hash-sig-keys"));
     // Optional slot-time override, for tests/sims that need a wider slot than the preset's.
@@ -346,6 +351,16 @@ pub fn build(b: *Builder) !void {
     });
     zeam_network.addImport("zig_libp2p", zig_libp2p_dep.module("zig_libp2p"));
 
+    // Optional ethp2p RS-broadcast transport. Only realize (fetch + import)
+    // the lazy `zig_ethp2p` dependency under `-Dethp2p=true`; otherwise the
+    // adapter selector compiles a stub that never imports it. (`build_options`
+    // is imported into `zeam_network` just below.)
+    if (ethp2p_enabled) {
+        if (b.lazyDependency("zig_ethp2p", .{ .target = target, .optimize = optimize })) |dep| {
+            zeam_network.addImport("zig_ethp2p", dep.module("zig_ethp2p"));
+        }
+    }
+
     // The publish-side forensic log line in v2 includes the build git SHA
     // so receivers across the fleet can correlate
     // broken-byte receipts back to the exact producer binary.

build.zig.zon

@@ -64,6 +64,14 @@
             .url = "https://github.com/blockblaz/zig-libp2p/archive/refs/tags/v0.2.89.tar.gz",
             .hash = "zig_libp2p-0.2.89-lil2hMltKgA7YdmHAS_eavWH0unkgsQhybBFEWqukws4",
         },
+        // Optional parallel RS-erasure-coded broadcast transport (ethp2p),
+        // wired only under `-Dethp2p=true`. Lazy so the default build neither
+        // fetches nor compiles it (nor its transitive zquic dependency).
+        .zig_ethp2p = .{
+            .url = "https://github.com/blockblaz/zig-ethp2p/archive/refs/tags/v0.1.3.tar.gz",
+            .hash = "zig_ethp2p-0.1.3-TcaWEpowCgCJdD_tCwd3SGQRxj9heTMCXQp_wGRFNBdY",
+            .lazy = true,
+        },
     },
     .paths = .{""},
 }

pkgs/cli/src/node.zig

@@ -68,6 +68,20 @@ pub const ValidatorAssignment = struct {
     }
 };
 
+/// Runtime toggle for the experimental ethp2p transport: truthy `ZEAM_ETHP2P`
+/// env var. Only consulted when the binary was built with `-Dethp2p=true`
+/// (see the comptime guard at the call site). Mirrors the `DEBUG_QUIC` /
+/// `ZEAM_STRESS_*` env-var convention; a CLI flag would trip zigcli's
+/// comptime branch quota.
+fn ethp2pRuntimeEnabled() bool {
+    const raw = std.c.getenv("ZEAM_ETHP2P") orelse return false;
+    const v = std.mem.trim(u8, std.mem.span(raw), " \t\n\r");
+    return std.mem.eql(u8, v, "1") or
+        std.ascii.eqlIgnoreCase(v, "true") or
+        std.ascii.eqlIgnoreCase(v, "yes") or
+        std.ascii.eqlIgnoreCase(v, "on");
+}
+
 pub const NodeOptions = struct {
     network_id: u32,
     node_key: []const u8,
@@ -656,6 +670,15 @@ pub const Node = struct {
             .aggregation_subnet_ids = options.aggregation_subnet_ids,
             .thread_pool = self.thread_pool,
             .chain_worker_enabled = options.chain_worker_enabled,
+            // Experimental ethp2p RS-broadcast: compiled in only under
+            // `-Dethp2p=true`, and then enabled at runtime only when the
+            // `ZEAM_ETHP2P` env var is truthy (zeam's env-var toggle
+            // convention; a CLI flag would trip zigcli's comptime branch
+            // quota). Off by default on both axes.
+            .ethp2p = if (comptime networks.ethp2p.enabled)
+                (if (ethp2pRuntimeEnabled()) networks.Ethp2pConfig{ .local_peer_id = "zeam-ethp2p" } else null)
+            else
+                null,
             .min_aggregation_inputs = options.min_aggregation_inputs,
             .max_aggregation_children = options.max_aggregation_children,
             .max_aggregations_per_tick = options.max_aggregations_per_tick,

pkgs/network/src/ethp2p.zig

@@ -0,0 +1,19 @@
+//! Compile-time selector for the optional ethp2p RS-broadcast adapter.
+//!
+//! Under `-Dethp2p=true` this resolves to the real `ethp2p_broadcast.zig`
+//! (which imports `zig_ethp2p`); otherwise it resolves to the inert
+//! `ethp2p_broadcast_stub.zig`, so the default build never touches the
+//! `zig_ethp2p` dependency.
+
+const build_options = @import("build_options");
+
+/// Whether the ethp2p adapter was compiled in.
+pub const enabled: bool = build_options.ethp2p;
+
+const impl = if (enabled)
+    @import("ethp2p_broadcast.zig")
+else
+    @import("ethp2p_broadcast_stub.zig");
+
+pub const Ethp2pBroadcast = impl.Ethp2pBroadcast;
+pub const Config = impl.Config;

pkgs/network/src/ethp2p_broadcast.zig

@@ -0,0 +1,169 @@
+//! Zeam-side adapter for the experimental **ethp2p** parallel RS-broadcast
+//! transport (mirrors ethlambda's `Ethp2pBroadcast`). It runs a separate QUIC
+//! network *alongside* libp2p/gossipsub: every outbound gossip publish is teed
+//! into a `zig_ethp2p` `BroadcastNode`, Reed-Solomon erasure-coded, and pushed
+//! to ethp2p peers; delivered messages are snappy-decompressed, SSZ-decoded,
+//! and reinjected into the node's gossip handler.
+//!
+//! Compiled only under `-Dethp2p=true` (see `ethp2p.zig`, which otherwise
+//! selects `ethp2p_broadcast_stub.zig`). Experimental and off by default:
+//! single-peer-correct outbound, no peer auth, no discovery-driven peering.
+
+const std = @import("std");
+const zig_ethp2p = @import("zig_ethp2p");
+const interface = @import("interface.zig");
+const config_mod = @import("ethp2p_config.zig");
+const snappyz = @import("snappyz");
+const ssz = @import("ssz");
+const types = @import("@zeam/types");
+
+const BroadcastNode = zig_ethp2p.node.broadcast_host.BroadcastNode;
+const BroadcastNodeConfig = zig_ethp2p.node.broadcast_host.BroadcastNodeConfig;
+const RsConfig = zig_ethp2p.layer.rs_init.RsConfig;
+const FullMessage = zig_ethp2p.node.broadcast_host.FullMessage;
+
+pub const Config = config_mod.Config;
+
+const log = std.log.scoped(.ethp2p);
+
+/// RS channels, named to match `GossipTopicKind` tag names so a channel id
+/// round-trips to/from `@tagName`.
+const channel_names = [_][]const u8{ "block", "aggregation", "attestation" };
+
+/// Snappy decode ceiling for delivered payloads.
+const max_decode: usize = 50 * 1024 * 1024;
+
+pub const Ethp2pBroadcast = struct {
+    allocator: std.mem.Allocator,
+    backend: interface.NetworkInterface,
+    node: *BroadcastNode,
+
+    pub fn start(
+        allocator: std.mem.Allocator,
+        backend: interface.NetworkInterface,
+        cfg: Config,
+    ) !*Ethp2pBroadcast {
+        const node = try BroadcastNode.init(allocator, .{
+            .local_peer_id = cfg.local_peer_id,
+            .listen_addr = cfg.listen_addr,
+            .server_certificate_pem_path = cfg.server_certificate_pem_path,
+            .server_private_key_pem_path = cfg.server_private_key_pem_path,
+            // Bounded so an unreachable static peer at startup can't hang the
+            // node for long.
+            .handshake_poll_rounds = 4000,
+        });
+        errdefer node.deinit();
+
+        const rs_cfg = RsConfig.default();
+        for (channel_names) |name| {
+            try node.addChannel(name, rs_cfg, cfg.sub_capacity);
+        }
+
+        // Best-effort dial of static peers; failures are logged, not fatal.
+        for (cfg.static_peers) |peer| {
+            node.connect(peer, cfg.server_name) catch |e| {
+                log.warn("ethp2p: dial {s} failed: {any}", .{ peer, e });
+            };
+        }
+
+        const self = try allocator.create(Ethp2pBroadcast);
+        self.* = .{ .allocator = allocator, .backend = backend, .node = node };
+
+        log.info(
+            "ethp2p broadcast started: peer_id={s} listen={?s} static_peers={d}",
+            .{ cfg.local_peer_id, cfg.listen_addr, cfg.static_peers.len },
+        );
+        return self;
+    }
+
+    pub fn deinit(self: *Ethp2pBroadcast) void {
+        self.node.deinit();
+        self.allocator.destroy(self);
+    }
+
+    /// Tee an outbound gossip message into the RS-broadcast network. Best-effort
+    /// — never propagates errors to the libp2p publish path.
+    pub fn publishGossip(self: *Ethp2pBroadcast, msg: *const interface.GossipMessage) void {
+        const ssz_bytes = msg.serialize(self.allocator) catch |e| {
+            log.debug("ethp2p tee: ssz serialize failed: {any}", .{e});
+            return;
+        };
+        defer self.allocator.free(ssz_bytes);
+
+        // message_id = hex(sha256(ssz_bytes)) — matches ethlambda's keying.
+        var digest: [32]u8 = undefined;
+        std.crypto.hash.sha2.Sha256.hash(ssz_bytes, &digest, .{});
+        const mid = std.fmt.bytesToHex(digest, .lower);
+
+        // RS payload is the snappy-compressed gossip body (same bytes libp2p
+        // would put on the wire).
+        const payload = snappyz.encode(self.allocator, ssz_bytes) catch |e| {
+            log.debug("ethp2p tee: snappy encode failed: {any}", .{e});
+            return;
+        };
+        defer self.allocator.free(payload);
+
+        const channel = @tagName(std.meta.activeTag(msg.*));
+        self.node.publish(channel, &mid, payload) catch |e| {
+            log.debug("ethp2p tee: publish failed channel={s}: {any}", .{ channel, e });
+            return;
+        };
+        log.debug(
+            "ethp2p tee: channel={s} mid={s} ssz={d} payload={d}",
+            .{ channel, &mid, ssz_bytes.len, payload.len },
+        );
+    }
+
+    /// Drive the RS-broadcast QUIC engine and reinject any reconstructed
+    /// messages back into the node's gossip handler.
+    pub fn tick(self: *Ethp2pBroadcast, now_ms: i64) void {
+        self.node.tick(now_ms) catch |e| {
+            log.debug("ethp2p tick: {any}", .{e});
+        };
+        for (channel_names) |name| {
+            while (self.node.poll(name)) |fm| {
+                self.deliver(name, fm);
+                self.node.freeMessage(fm);
+            }
+        }
+    }
+
+    /// Snappy-decompress + SSZ-decode a delivered RS message and hand it to the
+    /// gossip handler, exactly as the libp2p inbound path does. Best-effort.
+    fn deliver(self: *Ethp2pBroadcast, channel: []const u8, fm: FullMessage) void {
+        const kind = std.meta.stringToEnum(interface.GossipTopicKind, channel) orelse return;
+
+        const uncompressed = snappyz.decodeWithMax(self.allocator, fm.data, max_decode) catch |e| {
+            log.debug("ethp2p deliver: snappy decode failed channel={s}: {any}", .{ channel, e });
+            return;
+        };
+        defer self.allocator.free(uncompressed);
+
+        var message: interface.GossipMessage = switch (kind) {
+            .block => blk: {
+                var sb: types.SignedBlock = undefined;
+                ssz.deserialize(types.SignedBlock, uncompressed, &sb, self.allocator) catch return;
+                break :blk .{ .block = sb };
+            },
+            .aggregation => blk: {
+                var agg: types.SignedAggregatedAttestation = undefined;
+                ssz.deserialize(types.SignedAggregatedAttestation, uncompressed, &agg, self.allocator) catch return;
+                break :blk .{ .aggregation = agg };
+            },
+            .attestation => blk: {
+                var att: types.SignedAttestation = undefined;
+                ssz.deserialize(types.SignedAttestation, uncompressed, &att, self.allocator) catch return;
+                // The gossip subnet id lives in the libp2p topic, not the SSZ
+                // body, and the ethp2p channel does not carry it — reinject on
+                // subnet 0 (best-effort; cross-process attestation delivery is
+                // out of scope for this experimental adapter).
+                break :blk .{ .attestation = .{ .subnet_id = 0, .message = att } };
+            },
+        };
+        defer message.deinit();
+
+        self.backend.gossip.onGossipFn(self.backend.gossip.ptr, &message, "ethp2p") catch |e| {
+            log.debug("ethp2p deliver: onGossip failed channel={s}: {any}", .{ channel, e });
+        };
+    }
+};

pkgs/network/src/ethp2p_broadcast_stub.zig

@@ -0,0 +1,40 @@
+//! No-op stand-in for `ethp2p_broadcast.zig`, selected by `ethp2p.zig` when
+//! the build was not configured with `-Dethp2p=true`. It never imports
+//! `zig_ethp2p`, so the default build neither fetches nor compiles that
+//! dependency. The public API mirrors the real adapter so call sites compile
+//! unchanged; every method is an inert stub.
+
+const std = @import("std");
+const interface = @import("interface.zig");
+const config_mod = @import("ethp2p_config.zig");
+
+pub const Config = config_mod.Config;
+
+pub const Ethp2pBroadcast = struct {
+    pub fn start(
+        allocator: std.mem.Allocator,
+        backend: interface.NetworkInterface,
+        cfg: Config,
+    ) !*Ethp2pBroadcast {
+        _ = allocator;
+        _ = backend;
+        _ = cfg;
+        // The adapter is compiled out; enabling it at runtime without
+        // `-Dethp2p=true` is a configuration error.
+        return error.Ethp2pNotCompiledIn;
+    }
+
+    pub fn deinit(self: *Ethp2pBroadcast) void {
+        _ = self;
+    }
+
+    pub fn publishGossip(self: *Ethp2pBroadcast, msg: *const interface.GossipMessage) void {
+        _ = self;
+        _ = msg;
+    }
+
+    pub fn tick(self: *Ethp2pBroadcast, now_ms: i64) void {
+        _ = self;
+        _ = now_ms;
+    }
+};

pkgs/network/src/ethp2p_config.zig

@@ -0,0 +1,24 @@
+//! Shared config for the optional ethp2p RS-broadcast adapter. Kept in its own
+//! file so the real adapter and its stub share one type without either
+//! importing the other.
+
+/// Startup parameters for `Ethp2pBroadcast`. All peer/listen fields are
+/// optional: with none set the adapter runs dial-only with no peers, which is
+/// enough to tee gossip publishes into origin RS sessions (self-interop and
+/// cross-process delivery need `listen_addr` + `server_*_pem_path` on both
+/// ends and `static_peers` pointing at each other).
+pub const Config = struct {
+    /// Peer id announced in the BCAST handshake / used as the RS engine id.
+    local_peer_id: []const u8,
+    /// "host:port" to bind the QUIC listen (server) endpoint. Null → dial-only.
+    listen_addr: ?[]const u8 = null,
+    /// TLS server identity (PEM paths). Required when `listen_addr` is set.
+    server_certificate_pem_path: ?[]const u8 = null,
+    server_private_key_pem_path: ?[]const u8 = null,
+    /// Remote "host:port" peers to dial at startup (best-effort).
+    static_peers: []const []const u8 = &.{},
+    /// TLS server name (SNI) used on dials.
+    server_name: []const u8 = "127.0.0.1",
+    /// Per-channel delivery ring capacity.
+    sub_capacity: usize = 64,
+};

pkgs/network/src/lib.zig

@@ -44,6 +44,12 @@ pub const EthLibp2pParams = ethlibp2p.EthLibp2pParams;
 
 pub const gossip_codec = @import("./gossip_codec.zig");
 
+/// Optional ethp2p parallel RS-broadcast adapter (compile-time gated by
+/// `-Dethp2p`; a stub otherwise). See `ethp2p.zig`.
+pub const ethp2p = @import("./ethp2p.zig");
+pub const Ethp2pBroadcast = ethp2p.Ethp2pBroadcast;
+pub const Ethp2pConfig = ethp2p.Config;
+
 test "get tests" {
     @import("std").testing.refAllDecls(@This());
 }

pkgs/node/src/constants.zig

@@ -181,6 +181,18 @@ pub fn gossipStallThresholdMs() u64 {
 // batched `blocks_by_range` request, so anything beyond a few slots prefers range sync.
 pub const BLOCKS_BY_RANGE_SYNC_THRESHOLD: u64 = 4;
 
+// Hard ceiling on a `blocks_by_root` parent-walk catch-up. by-root fetches ONE
+// block per request, so the time to close a gap grows linearly with it: past a
+// few dozen slots the walk can never outrun block production and the node falls
+// permanently behind while storming its peers with retries.
+//
+// A devnet node wedged exactly this way: a peer-map miss was misreported as
+// "peer does not support blocks_by_range", which dropped a 13,429-slot catch-up
+// into by-root. Head froze for 5+ hours and the retry loop reached 2.4M requests.
+// Beyond this bound we now refuse to by-root and wait for a range-capable peer —
+// being visibly stuck is strictly better than a self-DoS that cannot converge.
+pub const MAX_BLOCKS_BY_ROOT_CATCHUP_GAP: u64 = 64;
+
 // Refuse to produce a block when our local head is this many wall-clock slots
 // behind, once the chain has reached its first justification. This prevents a
 // recovered/stale minority fork from minting current-slot blocks on an ancient