feat(networking): wire gossip→BN pipeline, gossipsub scoring feedback, outbound API

## Gossip → Beacon Node pipeline (issue ChainSafe#9)
- Add on_validated_message callback to EthGossipAdapter
  (setMessageHandler(ctx, callback) — called for every ACCEPT-validated message)
- validateDecoded() now delivers accepted messages to the BN processing pipeline
- Beacon node wires this to enqueue blocks/attestations for import

## Gossipsub ACCEPT/REJECT feedback (issue ChainSafe#11)
- handleMessage() now accepts from_peer: ?[]const u8 parameter
- For REJECT outcomes, calls gossipsub.router.recordInvalidMessage(peer, topic)
  which increments the P4 (invalid message delivery) score counter
- pollEvents() passes msg.from to handleMessage()
- Gossipsub can now penalize peers who send invalid messages

## Outbound request API (issue ChainSafe#10)
- Add decodeResponseChunks() + freeDecodedResponseChunks() to req_resp_encoding.zig
  to parse a complete response wire buffer into []DecodedResponseChunk
- handleOutbound() now decodes the response and calls ctx.on_response callback
  if present, passing parsed chunks to the caller (sync service, etc.)
- Export new types from root.zig (DecodedResponseChunk, decodeResponseChunks,
  freeDecodedResponseChunks)

🤖 Generated with AI assistance

Author: lodekeeper-z

src/networking/eth2_protocols.zig

@@ -142,9 +142,15 @@ fn makeProtocolHandler(
             };
         }
 
-        /// Handle an outbound request: encode SSZ and write to stream.
+        /// Handle an outbound request: encode SSZ, write to stream, decode response.
         ///
-        /// Expects `ctx` to have `ssz_payload: []const u8` (use `&.{}` for zero-body methods).
+        /// Expects `ctx` to optionally have:
+        /// - `ssz_payload: []const u8` — the request body (use `&.{}` for zero-body methods)
+        /// - `on_response: ?struct { ctx: *anyopaque, callback: *const fn(*anyopaque, []const ResponseChunk) void }`
+        ///   — called with parsed response chunks if present. This is how sync gets data back.
+        ///
+        /// The callback (if present) is invoked synchronously before `handleOutbound` returns.
+        /// Chunks are freed after the callback returns; the callback must copy what it needs.
         pub fn handleOutbound(self: *Self, io: Io, stream: anytype, ctx: anytype) !void {
             const ssz_bytes: []const u8 = if (@hasField(@TypeOf(ctx), "ssz_payload"))
                 ctx.ssz_payload
@@ -162,13 +168,38 @@ fn makeProtocolHandler(
                 return err;
             };
 
-            // Read response from peer (for request-response protocols).
+            // Read response from peer.
             const response_wire = readAllFromStream(self.allocator, io, stream, max_request_wire_bytes) catch |err| {
                 log.warn("{s} handleOutbound read response error: {}", .{ id, err });
                 return;
             };
             defer self.allocator.free(response_wire);
-            log.info("{s} handleOutbound: received {d} byte response", .{ id, response_wire.len });
+
+            // Decode the response wire bytes into typed response chunks.
+            // This makes the parsed data available to the caller (sync service, etc.)
+            // rather than silently discarding the response.
+            const chunks = req_resp_encoding.decodeResponseChunks(
+                self.allocator,
+                response_wire,
+                method.hasContextBytes(),
+                method.hasMultipleResponses(),
+            ) catch |err| {
+                log.warn("{s} handleOutbound decode response error: {}", .{ id, err });
+                return;
+            };
+            defer req_resp_encoding.freeDecodedResponseChunks(self.allocator, chunks);
+
+            log.info("{s} handleOutbound: received {d} byte response, {d} chunks", .{
+                id, response_wire.len, chunks.len,
+            });
+
+            // Deliver parsed chunks to the caller via on_response callback if provided.
+            // This is the outbound req/resp API — how sync service gets block/blob data.
+            if (@hasField(@TypeOf(ctx), "on_response")) {
+                if (ctx.on_response) |handler| {
+                    handler.callback(handler.ctx, chunks);
+                }
+            }
         }
 
         // Store method for diagnostics.

src/networking/eth_gossip.zig

@@ -58,6 +58,17 @@ pub const EthGossipAdapter = struct {
     fork_digest: [4]u8,
     /// Optional peer scorer for tracking validation outcomes.
     peer_scorer: ?*PeerScorer,
+    /// Optional callback invoked for each ACCEPT-validated gossip message.
+    ///
+    /// This is the gossip → beacon node pipeline entry point. When wired, the beacon
+    /// node (or BeaconProcessor) receives validated messages for import/processing.
+    ///
+    /// Signature: fn (ctx: *anyopaque, msg: DecodedGossipMessage) void
+    /// The `ctx` pointer is passed through opaquely; cast it to the concrete type.
+    on_validated_message: ?struct {
+        ctx: *anyopaque,
+        callback: *const fn (ctx: *anyopaque, msg: DecodedGossipMessage) void,
+    },
     /// Tracks which topics we have subscribed to (for cleanup).
     subscribed_topics: std.ArrayListUnmanaged([]const u8),
 
@@ -73,6 +84,7 @@ pub const EthGossipAdapter = struct {
             .validator = validator,
             .fork_digest = fork_digest,
             .peer_scorer = null,
+            .on_validated_message = null,
             .subscribed_topics = .empty,
         };
     }
@@ -82,6 +94,27 @@ pub const EthGossipAdapter = struct {
         self.peer_scorer = scorer;
     }
 
+    /// Wire a callback for delivering validated gossip messages to the beacon node.
+    ///
+    /// The callback is called for every ACCEPT-validated message, passing the typed
+    /// `DecodedGossipMessage` to the beacon node for processing (import block,
+    /// enqueue attestation, process exit, etc.).
+    ///
+    /// This is the gossip → beacon node pipeline. Without this, validated messages
+    /// are dropped after validation (correct for mesh scoring, wrong for sync).
+    ///
+    /// Usage (from beacon_node.zig):
+    /// ```zig
+    /// gossip_adapter.setMessageHandler(@ptrCast(self), &beaconNode_onGossipMessage);
+    /// ```
+    pub fn setMessageHandler(
+        self: *Self,
+        ctx: *anyopaque,
+        callback: *const fn (ctx: *anyopaque, msg: DecodedGossipMessage) void,
+    ) void {
+        self.on_validated_message = .{ .ctx = ctx, .callback = callback };
+    }
+
     pub fn deinit(self: *Self) void {
         for (self.subscribed_topics.items) |topic| {
             self.allocator.free(topic);
@@ -126,11 +159,15 @@ pub const EthGossipAdapter = struct {
     /// 3. Dispatch to per-topic validation
     /// 4. Return validation result + decoded message
     ///
+    /// `from_peer` is the peer that sent this message (for gossipsub scoring feedback).
+    /// Pass null if the sender peer ID is unavailable.
+    ///
     /// Called when gossipsub's `drainEvents()` yields a `message` event.
     pub fn handleMessage(
         self: *Self,
         topic: []const u8,
         data: []const u8,
+        from_peer: ?[]const u8,
     ) HandleMessageResult {
         // 1. Parse the topic string.
         const parsed_topic = gossip_topics.parseTopic(topic) orelse {
@@ -151,24 +188,59 @@ pub const EthGossipAdapter = struct {
         };
 
         // 3. Dispatch to per-topic validation.
-        const validation = self.validateDecoded(parsed_topic, decoded);
+        const validation = self.validateDecoded(parsed_topic, decoded, from_peer);
 
         return .{ .validation = validation, .decoded = decoded };
     }
 
     /// Validate a decoded gossip message using the per-topic validators.
+    ///
+    /// `from_peer` is used to report validation results back to gossipsub for
+    /// mesh scoring — rejected messages penalize the originating peer.
     fn validateDecoded(
         self: *Self,
         parsed_topic: GossipTopic,
         decoded: DecodedGossipMessage,
+        from_peer: ?[]const u8,
     ) ValidationResult {
         const result = self.dispatchValidation(parsed_topic, decoded);
 
-        // Update peer scoring if a scorer is wired.
+        // Update legacy peer scoring if a scorer is wired.
         if (self.peer_scorer) |scorer| {
             scorer.recordValidation(result);
         }
 
+        // Feed validation result back to gossipsub for mesh scoring.
+        // This is the gossipsub ACCEPT/REJECT feedback loop. Without it, gossipsub
+        // cannot penalize peers who send invalid messages, which is critical for
+        // spam resistance and mesh health.
+        if (from_peer) |peer| {
+            switch (result) {
+                .reject => {
+                    // recordInvalidMessage increments invalid_message_deliveries
+                    // for this peer on this topic, which reduces their P4 score.
+                    var topic_buf: [gossip_topics.MAX_TOPIC_LENGTH]u8 = undefined;
+                    const topic_str = gossip_topics.formatTopic(
+                        &topic_buf,
+                        self.fork_digest,
+                        parsed_topic.topic_type,
+                        parsed_topic.subnet_id,
+                    );
+                    self.gossipsub.router.recordInvalidMessage(peer, topic_str);
+                },
+                .accept, .ignore => {},
+            }
+        }
+
+        // Deliver accepted messages to the beacon node processing pipeline.
+        // This is the gossip → BN handoff. Without this, validated messages are
+        // silently discarded and never imported into the chain.
+        if (result == .accept) {
+            if (self.on_validated_message) |handler| {
+                handler.callback(handler.ctx, decoded);
+            }
+        }
+
         return result;
     }
 
@@ -286,7 +358,7 @@ pub const EthGossipAdapter = struct {
         for (events) |event| {
             switch (event) {
                 .message => |msg| {
-                    const handle_result = self.handleMessage(msg.topic, msg.data);
+                    const handle_result = self.handleMessage(msg.topic, msg.data, msg.from);
                     const topic_copy = try self.allocator.dupe(u8, msg.topic);
                     try results.append(self.allocator, .{
                         .topic = topic_copy,

src/networking/req_resp_encoding.zig

@@ -487,3 +487,53 @@ test "response chunk roundtrip with context bytes and BeaconBlocksByRange" {
     try testing.expectEqual(request.count, decoded_req.count);
     try testing.expectEqual(request.step, decoded_req.step);
 }
+
+/// Decode all response chunks from a complete response wire buffer.
+///
+/// Iterates through the wire bytes, decoding one chunk at a time until all
+/// bytes are consumed. Stops early on the first non-success chunk.
+///
+/// Caller owns the returned slice and each `ssz_bytes` field within it.
+/// Use `freeResponseChunks` to release all memory.
+///
+/// This is used by the outbound request API to parse responses from peers.
+pub fn decodeResponseChunks(
+    allocator: std.mem.Allocator,
+    wire_bytes: []const u8,
+    has_context_bytes: bool,
+    has_multiple_responses: bool,
+) ![]DecodedResponseChunk {
+    var chunks: std.ArrayListUnmanaged(DecodedResponseChunk) = .empty;
+    errdefer {
+        for (chunks.items) |chunk| {
+            allocator.free(chunk.ssz_bytes);
+        }
+        chunks.deinit(allocator);
+    }
+
+    var offset: usize = 0;
+    while (offset < wire_bytes.len) {
+        const chunk = try decodeResponseChunk(allocator, wire_bytes[offset..], has_context_bytes);
+        offset += chunk.bytes_consumed;
+        try chunks.append(allocator, chunk);
+
+        // Single-response protocols (Status, Goodbye, Ping, Metadata) return
+        // exactly one chunk. Stop after reading it.
+        if (!has_multiple_responses) break;
+
+        // Error responses terminate the stream.
+        if (!chunk.result.isSuccess()) break;
+    }
+
+    return chunks.toOwnedSlice(allocator);
+}
+
+/// Free a slice of response chunks allocated by `decodeResponseChunks`.
+///
+/// Frees both the ssz_bytes within each chunk and the slice itself.
+pub fn freeDecodedResponseChunks(allocator: std.mem.Allocator, chunks: []DecodedResponseChunk) void {
+    for (chunks) |chunk| {
+        allocator.free(chunk.ssz_bytes);
+    }
+    allocator.free(chunks);
+}

src/networking/root.zig

@@ -29,6 +29,9 @@ pub const formatProtocolId = protocol.formatProtocolId;
 pub const decodeRequest = req_resp_encoding.decodeRequest;
 pub const encodeResponseChunk = req_resp_encoding.encodeResponseChunk;
 pub const decodeResponseChunk = req_resp_encoding.decodeResponseChunk;
+pub const decodeResponseChunks = req_resp_encoding.decodeResponseChunks;
+pub const freeDecodedResponseChunks = req_resp_encoding.freeDecodedResponseChunks;
+pub const DecodedResponseChunk = req_resp_encoding.DecodedResponseChunk;
 
 // Req/resp handler re-exports.
 pub const ReqRespContext = req_resp_handler.ReqRespContext;