broadcast: session reconstruction signaling + disposal (SESS reset 0x01) (#71)

Adds the session lifecycle state machine, aligned with ethp2p
`broadcast/session.go`:
  - `SessionStage` (consuming/decoding/reconstructed/origin) with ordering
    semantics (`isDecoded` == stage >= reconstructed),
  - `sessCodeReconstructed = 0x01` — the QUIC app error code a peer's SESS
    stream is reset with once reconstructed (`broadcast/peer_in.go`),
  - `signalReconstructed`, per-peer `markPeerCompleted`/`dropPeer`, and
    `maybeDispose` (not-decoded → all peers dropped; decoded → all
    attached peers completed; peerless sessions left for TTL cleanup),
  - `ChannelRs.disposeSession` — removes + frees the session and emits
    `OnSessionDisposed` (completing another Observer callback from #58).

`sessionDecode` now calls `signalReconstructed` (no-op for origin).
Origin sessions start at `.origin`, relay sessions at `.consuming`.

The actual per-peer SESS-stream reset with 0x01 is emitted by the QUIC
transport when it observes reconstruction — a transport wiring follow-up,
since `SessionRs` is transport-agnostic.

Tests: stage ordering + reset-code constant, origin disposal gated on
peer completion (with `OnSessionDisposed` emit), and relay
consuming→reconstructed + drop-based disposal — all leak-checked.
README implementation-status table updated.

Closes #62.

Author: ch4r10t33r

README.md

@@ -17,6 +17,7 @@ Zig helpers for the wire formats of **[ethp2p](https://github.com/ethp2p/ethp2p)
 | `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` |
 | `FullMessage` + channel `Subscribe` / decoded-message delivery (bounded FIFO `Subscription`, drop-on-full; `AlreadySubscribed` / `UnbufferedSubscription`) | [`broadcast/channel.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/channel.go) `Subscribe` / `deliverMessage`, [`broadcast/types.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/types.go) `FullMessage` | `broadcast.channel_rs` (`Subscription`), `layer.broadcast_types` |
+| Session lifecycle: `SessionStage` machine, `signalReconstructed`, per-peer complete/drop tracking, `maybeDispose`, `disposeSession` (emits `OnSessionDisposed`); `sessCodeReconstructed = 0x01` SESS-reset code (per-peer stream reset itself is a transport follow-up) | [`broadcast/session.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/session.go), [`broadcast/peer_in.go`](https://github.com/ethp2p/ethp2p/blob/main/broadcast/peer_in.go) `sessCodeReconstructed` | `broadcast.session_rs`, `broadcast.channel_rs` |
 | `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` |

src/broadcast/channel_rs.zig

@@ -183,6 +183,12 @@ pub const ChannelRs = struct {
         }
         const stats = try self.allocator.alloc(broadcast_types.PeerSessionStats, n);
         errdefer self.allocator.free(stats);
+        const completed = try self.allocator.alloc(bool, n);
+        errdefer self.allocator.free(completed);
+        @memset(completed, false);
+        const dropped = try self.allocator.alloc(bool, n);
+        errdefer self.allocator.free(dropped);
+        @memset(dropped, false);
 
         while (populated < n) : (populated += 1) {
             member_ids[populated] = try self.allocator.dupe(u8, self.members.items[populated]);
@@ -201,6 +207,10 @@ pub const ChannelRs = struct {
             .strategy = strat.?,
             .member_ids = member_ids,
             .stats = stats,
+            .stage = .origin,
+            .completed = completed,
+            .dropped = dropped,
+            .ever_had_peers = n > 0,
         };
         strat = null;
 
@@ -227,6 +237,12 @@ pub const ChannelRs = struct {
         }
         const stats = try self.allocator.alloc(broadcast_types.PeerSessionStats, n);
         errdefer self.allocator.free(stats);
+        const completed = try self.allocator.alloc(bool, n);
+        errdefer self.allocator.free(completed);
+        @memset(completed, false);
+        const dropped = try self.allocator.alloc(bool, n);
+        errdefer self.allocator.free(dropped);
+        @memset(dropped, false);
 
         while (populated < n) : (populated += 1) {
             member_ids[populated] = try self.allocator.dupe(u8, self.members.items[populated]);
@@ -245,6 +261,10 @@ pub const ChannelRs = struct {
             .strategy = strat,
             .member_ids = member_ids,
             .stats = stats,
+            .stage = .consuming,
+            .completed = completed,
+            .dropped = dropped,
+            .ever_had_peers = n > 0,
         };
 
         try self.sessions.put(self.allocator, mid, sess);
@@ -271,13 +291,27 @@ pub const ChannelRs = struct {
         const slot = self.sessions.getPtr(message_id) orelse return error.InvalidMessage;
         const out = try slot.*.strategy.decode();
         errdefer self.allocator.free(out);
+        // Relay session reconstructed (no-op for origin); mirrors Go
+        // `signalReconstructed` after a successful decode.
+        slot.*.signalReconstructed();
         // 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);
         try self.deliverMessage(message_id, out);
         return out;
     }
 
+    /// Remove `message_id`'s session and free it, emitting `OnSessionDisposed`.
+    /// Mirrors ethp2p `Channel.disposeSession`; no-op if there is no session.
+    pub fn disposeSession(self: *ChannelRs, message_id: []const u8, reason: []const u8) void {
+        if (self.sessions.fetchRemove(message_id)) |kv| {
+            self.engine.config.observer.sessionDisposed(self.id, kv.key, reason);
+            kv.value.deinit();
+            self.allocator.destroy(kv.value);
+            self.allocator.free(kv.key);
+        }
+    }
+
     /// After a successful decode, clears engine `DedupRegistry` keys for this `(channel_id, message_id)`
     /// when `EngineConfig.enable_cross_session_dedup` is set (no-op otherwise).
     pub fn sessionDecodeClearEngineDedup(self: *ChannelRs, message_id: []const u8) ![]u8 {
@@ -493,6 +527,63 @@ test "subscriber drops decoded messages when full" {
     try std.testing.expectEqual(@as(usize, 1), sub.dropped());
 }
 
+const session_rs_mod = @import("session_rs.zig");
+
+test "origin session disposes once its peer completes; disposeSession emits" {
+    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 ch = try eng.attachChannelRs("topic", sub_test_cfg);
+    try ch.addMember("p1");
+
+    const payload = [_]u8{ 'h', 'i', '!', '!', '!' };
+    try ch.publish("m1", &payload);
+
+    const sess = ch.sessions.get("m1").?;
+    try std.testing.expectEqual(session_rs_mod.SessionStage.origin, sess.stage);
+    // Decoded (origin) but the peer has not completed → not disposable yet.
+    try std.testing.expect(!sess.maybeDispose());
+    sess.markPeerCompleted("p1");
+    try std.testing.expect(sess.maybeDispose());
+
+    ch.disposeSession("m1", "reconstructed");
+    try std.testing.expect(ch.sessions.get("m1") == null);
+    try std.testing.expectEqual(@as(usize, 1), rec.session_disposed);
+}
+
+test "relay session: consuming stage, reconstruct signal, drop-based disposal" {
+    const gpa = std.testing.allocator;
+    var eng = try Engine.init(gpa, "local", .{});
+    defer eng.deinit();
+    const ch = try eng.attachChannelRs("topic", sub_test_cfg);
+    try ch.addMember("peerA");
+    try ch.addMember("peerB");
+
+    const payload = [_]u8{ 9, 8, 7 };
+    var origin = try RsStrategy.newOrigin(gpa, sub_test_cfg, &payload);
+    defer origin.deinit();
+    try ch.attachRelaySession("m1", &origin.preamble);
+
+    const sess = ch.sessions.get("m1").?;
+    try std.testing.expectEqual(session_rs_mod.SessionStage.consuming, sess.stage);
+
+    // Not decoded: disposes only when every peer is dropped.
+    try std.testing.expect(!sess.maybeDispose());
+    sess.dropPeer("peerA");
+    try std.testing.expect(!sess.maybeDispose());
+    sess.dropPeer("peerB");
+    try std.testing.expect(sess.maybeDispose());
+
+    // Reconstruction moves it to the decoded branch.
+    sess.signalReconstructed();
+    try std.testing.expectEqual(session_rs_mod.SessionStage.reconstructed, sess.stage);
+    try std.testing.expect(sess.maybeDispose());
+
+    ch.disposeSession("m1", "ttl_expired");
+    try std.testing.expect(ch.sessions.get("m1") == null);
+}
+
 test "relay ingest registry blocks second claim same chunk index" {
     const gpa = std.testing.allocator;
     var eng = try Engine.init(gpa, "local", .{});

src/broadcast/observer.zig

@@ -144,6 +144,7 @@ const testing = std.testing;
 pub const Recording = struct {
     session_started: usize = 0,
     session_decoded: usize = 0,
+    session_disposed: usize = 0,
     chunk_rcvd: usize = 0,
     peer_subscribed: usize = 0,
     peer_unsubscribed: usize = 0,
@@ -164,7 +165,7 @@ pub const Recording = struct {
         .onPeerGone = noop_impl.peerGone,
         .onSessionStarted = onSessionStarted,
         .onSessionDecoded = onSessionDecoded,
-        .onSessionDisposed = noop_impl.sessionDisposed,
+        .onSessionDisposed = onSessionDisposed,
         .onChunkSent = noop_impl.chunkSent,
         .onChunkRcvd = onChunkRcvd,
         .onChunkError = noop_impl.chunkError,
@@ -190,6 +191,9 @@ pub const Recording = struct {
     fn onSessionDecoded(ctx: *anyopaque, _: []const u8, _: []const u8, _: i64) void {
         cast(ctx).session_decoded += 1;
     }
+    fn onSessionDisposed(ctx: *anyopaque, _: []const u8, _: []const u8, _: []const u8) void {
+        cast(ctx).session_disposed += 1;
+    }
     fn onChunkRcvd(ctx: *anyopaque, _: []const u8, _: []const u8, _: []const u8, verdict: Verdict) void {
         const self = cast(ctx);
         self.chunk_rcvd += 1;

src/broadcast/session_rs.zig

@@ -17,19 +17,92 @@ pub const SendRsChunkFn = *const fn (
     payload: []const u8,
 ) anyerror!void;
 
+/// QUIC application error code a peer's SESS stream is reset with once the
+/// session is reconstructed, telling the remote to stop sending chunks.
+/// Mirrors ethp2p `sessCodeReconstructed` (`broadcast/peer_in.go`).
+pub const sess_code_reconstructed: u64 = 0x01;
+
+/// Linear session lifecycle (ethp2p `sessionStage`). Ordering is meaningful:
+/// `stage >= .reconstructed` means "decoded" (a relay that reconstructed, or an
+/// origin that always had the payload — `.origin` sorts highest).
+pub const SessionStage = enum(u8) {
+    consuming = 0, // relay, accepting chunks
+    decoding = 1, // relay, decode running
+    reconstructed = 2, // relay, decode succeeded
+    origin = 3, // origin, has the payload from the start
+
+    /// Whether the session holds the full message (origin or reconstructed relay).
+    pub fn isDecoded(self: SessionStage) bool {
+        return @intFromEnum(self) >= @intFromEnum(SessionStage.reconstructed);
+    }
+};
+
 pub const SessionRs = struct {
     allocator: Allocator,
     /// Owned by the parent `ChannelRs` map key; not freed here.
     message_id: []const u8,
     strategy: RsStrategy,
     member_ids: [][]u8,
     stats: []broadcast_types.PeerSessionStats,
+    stage: SessionStage = .consuming,
+    /// Per-member flags, parallel to `member_ids` (Go tracks these in `peers`).
+    completed: []bool = &.{},
+    dropped: []bool = &.{},
+    /// Whether any peer was ever attached (Go `everHadPeers`); disposal is
+    /// gated on this so peerless sessions are left for TTL cleanup.
+    ever_had_peers: bool = false,
 
     pub fn deinit(self: *SessionRs) void {
         self.strategy.deinit();
         for (self.member_ids) |m| self.allocator.free(m);
         self.allocator.free(self.member_ids);
         self.allocator.free(self.stats);
+        self.allocator.free(self.completed);
+        self.allocator.free(self.dropped);
+    }
+
+    /// Mark the session reconstructed after a successful decode
+    /// (ethp2p `signalReconstructed`). No-op for origin sessions.
+    pub fn signalReconstructed(self: *SessionRs) void {
+        if (self.stage == .origin) return;
+        self.stage = .reconstructed;
+    }
+
+    fn peerIndex(self: *const SessionRs, peer_id: []const u8) ?usize {
+        for (self.member_ids, 0..) |m, i| {
+            if (std.mem.eql(u8, m, peer_id)) return i;
+        }
+        return null;
+    }
+
+    /// Mark a peer as having completed this session (ethp2p `handlePeerCompleted`).
+    pub fn markPeerCompleted(self: *SessionRs, peer_id: []const u8) void {
+        if (self.peerIndex(peer_id)) |i| self.completed[i] = true;
+    }
+
+    /// Mark a peer as dropped (disconnected). Dropped peers are excluded from
+    /// disposal accounting (Go removes them from the `peers` map).
+    pub fn dropPeer(self: *SessionRs, peer_id: []const u8) void {
+        if (self.peerIndex(peer_id)) |i| self.dropped[i] = true;
+    }
+
+    /// Whether the session has no remaining work and can be disposed
+    /// (ethp2p `maybeDispose`):
+    ///   - not decoded: dispose once every peer has been dropped,
+    ///   - decoded: dispose once every still-attached peer has completed.
+    /// Peerless sessions never auto-dispose (left for TTL cleanup).
+    pub fn maybeDispose(self: *const SessionRs) bool {
+        if (!self.ever_had_peers) return false;
+        if (!self.stage.isDecoded()) {
+            for (self.dropped) |d| {
+                if (!d) return false; // a live peer still needs chunks from us
+            }
+            return true;
+        }
+        for (self.completed, self.dropped) |c, d| {
+            if (!d and !c) return false; // an attached peer has not finished
+        }
+        return true;
     }
 
     /// One scheduling step: emit dispatches, then mark sends successful (simulates transport ACK).
@@ -78,3 +151,11 @@ pub const SessionRs = struct {
         return total;
     }
 };
+
+test "session stage ordering and reconstructed reset code" {
+    try std.testing.expectEqual(@as(u64, 0x01), sess_code_reconstructed);
+    try std.testing.expect(!SessionStage.consuming.isDecoded());
+    try std.testing.expect(!SessionStage.decoding.isDecoded());
+    try std.testing.expect(SessionStage.reconstructed.isDecoded());
+    try std.testing.expect(SessionStage.origin.isDecoded());
+}