broadcast: implement Observer interface + NoOpObserver + SessionRole

README.md

@@ -14,7 +14,8 @@ Zig helpers for the wire formats of **[ethp2p](https://github.com/ethp2p/ethp2p)
 | CHUNK uni-stream | [`broadcast/peer_ctrl.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/peer_ctrl.go) `doSendChunk`, [`peer_in.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/peer_in.go) `processChunk` | `wire.chunk_stream` |
 | RS `Preamble` / `ChunkIdent` | [`broadcast/rs/pb`](https://github.com/ethp2p/ethp2p/tree/main/broadcast/rs/pb), [`broadcast/rs/types.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/rs/types.go) | `wire.rs`, `writeRsShardChunk` |
 | Protocol version constant | [`broadcast/types.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/types.go) `ProtocolV1` | `wire.constants.protocol_v1` |
-| `Verdict` / `ChunkHandle` | [`broadcast/types.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/types.go) | `layer.broadcast_types` |
+| `Verdict` / `ChunkHandle` / `SessionRole` | [`broadcast/types.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/types.go), [`broadcast/observer.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/observer.go) | `layer.broadcast_types` |
+| Broadcast event `Observer` interface + `NoOpObserver` (15 callbacks, type-erased vtable) — emitted for channel-attach, session start/decode, chunk receive, peer subscribe/unsubscribe (remaining callbacks land with their subsystems) | [`broadcast/observer.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/observer.go) | `broadcast.observer` |
 | `DedupCancel` + `DedupGroup` token (session hook) | session → strategy `takeChunk` | `layer.broadcast_types`, `layer.dedup` |
 | Engine-wide dedup registry (`channel` + `message` + chunk index) | multi-peer ingest dedup | `layer.dedup_registry`, `Engine.enable_cross_session_dedup` |
 | Verify result FIFO (single-threaded `Verified()` shim) | async verify channels | `layer.verify_queue` |
@@ -39,6 +40,7 @@ Zig helpers for the wire formats of **[ethp2p](https://github.com/ethp2p/ethp2p)
 | `PartialMessagesExtension` (nested in `RPC.partial`) | libp2p `rpc.proto` field 10 body | `encodePartialMessagesExtension`, `decodePartialMessagesExtensionOwned` |
 | Unsigned-varint length prefix before `RPC` body | common libp2p framing | `encodeRpcLengthPrefixed`, `decodeRpcLengthPrefixedPrefix` |
 | In-process duplex for length-prefixed `RPC` (simnet-style, no TCP/QUIC) | pair of `Endpoint`s over bounded byte queues | `sim.gossipsub_rpc_host` (`Link`, `Endpoint.sendRpc` / `recvRpcOwned`) |
+| Broadcast trace writer (`.bctrace` NDJSON: header / event tuples / 100 ms footer time index; golden bytes from the Go writer) | [`sim/trace_writer.go`](https://github.com/ethp2p/ethp2p/blob/main/sim/trace_writer.go) | `sim.trace_writer` |
 | QUIC transport + UNI stream alignment | [`sim/host.go`](https://github.com/ethp2p/ethp2p/blob/main/sim/host.go), `peer.go`, `peer_ctrl.go`, `peer_in.go` | `transport.eth_ec_quic`: IETF QUIC, TLS 1.3, ALPN `eth-ec-broadcast`, **unidirectional** BCAST/SESS/CHUNK streams matching `OpenUniStream`/`AcceptUniStream`; `PeerConn` + `broadcast.engine_quic` (`EngineQuicHost`) wire inbound SESS/CHUNK into `Engine` / `ChannelRs`. See [QUIC transport](#quic-transport-zquic). |
 | **Still open** | — | [Pending work](#pending-work) |
 
@@ -96,7 +98,7 @@ All ethp2p application protocols use UNI streams — both peers independently op
 
 ## Requirements
 
-- Zig **0.15.1** or newer.
+- Zig **0.16.0** or newer. (The `std.Io` reorg — `std.net`, the `std.io` reader/writer, `std.crypto.random`, and `std.Thread.Mutex`/`Pool` — is bridged by the dependency-free `src/compat.zig` shim.)
 
 ## Build
 

src/broadcast/channel_rs.zig

@@ -60,13 +60,15 @@ pub const ChannelRs = struct {
         const dup = try self.allocator.dupe(u8, peer_id);
         errdefer self.allocator.free(dup);
         try self.members.append(self.allocator, dup);
+        self.engine.config.observer.peerSubscribed(peer_id, self.id);
     }
 
     pub fn removeMember(self: *ChannelRs, peer_id: []const u8) void {
         for (self.members.items, 0..) |m, i| {
             if (std.mem.eql(u8, m, peer_id)) {
                 self.allocator.free(m);
                 _ = self.members.swapRemove(i);
+                self.engine.config.observer.peerUnsubscribed(peer_id, self.id);
                 return;
             }
         }
@@ -113,6 +115,7 @@ pub const ChannelRs = struct {
         strat = null;
 
         try self.sessions.put(self.allocator, mid, sess);
+        self.engine.config.observer.sessionStarted(self.id, mid, .origin);
     }
 
     /// Relay-side session for an existing preamble (same members as `publish`).
@@ -155,6 +158,7 @@ pub const ChannelRs = struct {
         };
 
         try self.sessions.put(self.allocator, mid, sess);
+        self.engine.config.observer.sessionStarted(self.id, mid, .relay);
     }
 
     pub fn sessionDrainOutbound(self: *ChannelRs, message_id: []const u8) !usize {
@@ -175,7 +179,11 @@ pub const ChannelRs = struct {
 
     pub fn sessionDecode(self: *ChannelRs, message_id: []const u8) ![]u8 {
         const slot = self.sessions.getPtr(message_id) orelse return error.InvalidMessage;
-        return slot.*.strategy.decode();
+        const out = try slot.*.strategy.decode();
+        // Latency is not tracked in the Zig port yet (no per-session start clock);
+        // emit 0 for now — Go passes `time.Since(session.start)`.
+        self.engine.config.observer.sessionDecoded(self.id, message_id, 0);
+        return out;
     }
 
     /// After a successful decode, clears engine `DedupRegistry` keys for this `(channel_id, message_id)`
@@ -204,13 +212,15 @@ pub const ChannelRs = struct {
         dedup: ?*broadcast_types.DedupCancel,
     ) (Allocator.Error || errors.Error)!broadcast_types.ChunkIngestResult {
         const strat = self.sessionStrategy(message_id) orelse return error.InvalidMessage;
-        if (registry) |reg| {
-            const first = try reg.claim(self.allocator, self.id, message_id, chunk_id.index);
-            if (!first) {
-                return .{ .verdict = .redundant, .complete = false };
+        const result: broadcast_types.ChunkIngestResult = blk: {
+            if (registry) |reg| {
+                const first = try reg.claim(self.allocator, self.id, message_id, chunk_id.index);
+                if (!first) break :blk .{ .verdict = .redundant, .complete = false };
             }
-        }
-        return strat.takeChunk(peer, chunk_id, data, dedup);
+            break :blk try strat.takeChunk(peer, chunk_id, data, dedup);
+        };
+        self.engine.config.observer.chunkRcvd(peer, self.id, message_id, result.verdict);
+        return result;
     }
 
     /// `relayIngestChunk` using the engine-owned registry when cross-session dedup is enabled; otherwise `null`.
@@ -238,6 +248,7 @@ pub const ChannelRs = struct {
         const strat = self.sessionStrategy(message_id) orelse return error.InvalidMessage;
         const v = strat.verifyChunk(chunk_id, data);
         if (v != .accepted) {
+            self.engine.config.observer.chunkRcvd(peer, self.id, message_id, v);
             return .{ .verdict = v, .complete = false };
         }
         return self.relayIngestChunk(registry, message_id, peer, chunk_id, data, dedup);
@@ -284,6 +295,36 @@ test "channel publish and drain to one member" {
     try std.testing.expectEqualSlices(u8, &payload, decoded);
 }
 
+test "observer receives channel/peer/session events" {
+    const gpa = std.testing.allocator;
+    var rec: @import("observer.zig").Recording = .{};
+
+    var eng = try Engine.init(gpa, "local", .{ .observer = rec.observer() });
+    defer eng.deinit();
+
+    const cfg = RsConfig{
+        .data_shards = 4,
+        .parity_shards = 2,
+        .chunk_len = 0,
+        .bitmap_threshold = 0,
+        .forward_multiplier = 4,
+        .disable_bitmap = false,
+    };
+
+    const ch = try eng.attachChannelRs("topic", cfg);
+    try ch.addMember("p1");
+    const payload = [_]u8{ 'h', 'e', 'l', 'l', 'o' };
+    try ch.publish("m1", &payload);
+    const decoded = try ch.sessionDecode("m1");
+    defer gpa.free(decoded);
+
+    try std.testing.expectEqual(@as(usize, 1), rec.channel_attached);
+    try std.testing.expectEqual(@as(usize, 1), rec.peer_subscribed);
+    try std.testing.expectEqual(@as(usize, 1), rec.session_started);
+    try std.testing.expectEqual(@import("../layer/broadcast_types.zig").SessionRole.origin, rec.last_role.?);
+    try std.testing.expectEqual(@as(usize, 1), rec.session_decoded);
+}
+
 test "relay ingest registry blocks second claim same chunk index" {
     const gpa = std.testing.allocator;
     var eng = try Engine.init(gpa, "local", .{});

src/broadcast/engine.zig

@@ -11,7 +11,7 @@ const Allocator = std.mem.Allocator;
 pub const Error = errors.Error;
 
 pub const EngineConfig = struct {
-    observer: observer_mod.Observer = .{},
+    observer: observer_mod.Observer = observer_mod.noop,
     /// When set, `Engine` owns a `DedupRegistry` for `relayIngestChunk`-style helpers.
     enable_cross_session_dedup: bool = false,
 };
@@ -75,6 +75,7 @@ pub const Engine = struct {
         errdefer self.allocator.destroy(ch);
         ch.* = try ChannelRs.init(self.allocator, self, key, cfg);
         try self.channels.put(self.allocator, key, ch);
+        self.config.observer.channelAttached(key, null);
         return ch;
     }