diff --git a/src/quic/connection.zig b/src/quic/connection.zig index 9dbc52d..67d35dd 100644 --- a/src/quic/connection.zig +++ b/src/quic/connection.zig @@ -1544,6 +1544,9 @@ pub const Connection = struct { if (stream_mod.isBidi(sf.stream_id)) { if (self.streams.getStream(sf.stream_id)) |s| { try s.send.onAck(sf.offset, sf.length); + if (s.closed_for_gc and s.send.retransmit_count == 0 and !s.send.hasUnackedData()) { + self.streams.queueDisposal(s.stream_id); + } } } else { if (self.streams.send_streams.get(sf.stream_id)) |s| { @@ -1677,6 +1680,9 @@ pub const Connection = struct { if (stream_mod.isBidi(sf.stream_id)) { if (self.streams.getStream(sf.stream_id)) |s| { try s.send.onAck(sf.offset, sf.length); + if (s.closed_for_gc and s.send.retransmit_count == 0 and !s.send.hasUnackedData()) { + self.streams.queueDisposal(s.stream_id); + } } } else { if (self.streams.send_streams.get(sf.stream_id)) |s| { diff --git a/src/quic/stream.zig b/src/quic/stream.zig index 79d4145..3e2c4c3 100644 --- a/src/quic/stream.zig +++ b/src/quic/stream.zig @@ -2166,3 +2166,104 @@ test "StreamsMap: setSendOrder dynamically changes scheduling" { try testing.expectEqual(@as(u64, 0), out[0].stream_id); try testing.expectEqual(@as(u64, 4), out[1].stream_id); } + +// ── Stream lifecycle / disposal tests ────────────────────────────────── + +test "collectClosedStreams marks fully-closed bidi streams" { + var sm = StreamsMap.init(testing.allocator, false); + defer sm.deinit(); + sm.setMaxStreams(10, 10); + + const s = try sm.openBidiStream(); + try testing.expectEqual(@as(u64, 1), sm.open_bidi_streams); + + // Simulate both directions done: our FIN sent, peer FIN received + s.send.fin_sent = true; + s.recv.fin_received = true; + + // openBidiStream doesn't set needs_gc_scan (only getOrCreateStream does), so set it manually + sm.needs_gc_scan = true; + sm.collectClosedStreams(); + + try testing.expect(s.closed_for_gc); + // closeStream decremented open counter (locally-initiated, consumed not incremented) + try testing.expectEqual(@as(u64, 0), sm.open_bidi_streams); + try testing.expectEqual(@as(u64, 0), sm.consumed_bidi_streams); +} + +test "queueDisposal + drainDisposalQueue lifecycle" { + var sm = StreamsMap.init(testing.allocator, false); + defer sm.deinit(); + sm.setMaxStreams(10, 10); + + const s = try sm.openBidiStream(); + const sid = s.stream_id; + try testing.expect(sm.streams.get(sid) != null); + + // Queue — stream still in hashmap before drain + sm.queueDisposal(sid); + try testing.expect(sm.streams.get(sid) != null); + + // Drain — stream removed, memory freed + sm.drainDisposalQueue(); + try testing.expect(sm.streams.get(sid) == null); +} + +test "collectClosedStreams defers disposal when hasUnackedData" { + var sm = StreamsMap.init(testing.allocator, false); + defer sm.deinit(); + sm.setMaxStreams(10, 10); + + const s = try sm.openBidiStream(); + try s.send.writeData("x" ** 1000); + s.send.send_offset = 1000; // all data "sent" + s.send.fin_sent = true; + s.recv.fin_received = true; + + // ack_offset is 0, so hasUnackedData returns true + try testing.expect(s.send.hasUnackedData()); + + // collectClosedStreams marks closed_for_gc but keeps stream in map (no disposal) + sm.needs_gc_scan = true; + sm.collectClosedStreams(); + try testing.expect(s.closed_for_gc); + try testing.expect(sm.streams.get(s.stream_id) != null); + + // Simulate ACK for all data — advances ack_offset to 1000 + try s.send.onAck(0, 1000); + try testing.expect(!s.send.hasUnackedData()); +} + +test "disposal guard: closed_for_gc + fully-acked -> disposed" { + var sm = StreamsMap.init(testing.allocator, false); + defer sm.deinit(); + sm.setMaxStreams(10, 10); + + const s = try sm.openBidiStream(); + try s.send.writeData("x" ** 100); + s.send.send_offset = 100; + s.send.fin_sent = true; + s.recv.fin_received = true; + + // Mark closed_for_gc via collectClosedStreams + sm.needs_gc_scan = true; + sm.collectClosedStreams(); + try testing.expect(s.closed_for_gc); + + // Fully ACK all sent data (fin_queued not set, so hasUnackedData becomes false) + try s.send.onAck(0, 100); + try testing.expect(!s.send.hasUnackedData()); + try testing.expectEqual(@as(u8, 0), s.send.retransmit_count); + + // Same guard used by the onAck disposal check and the STREAM+FIN path + // closed_for_gc && retransmit_count == 0 && !hasUnackedData() + try testing.expect(s.closed_for_gc); + try testing.expectEqual(@as(u8, 0), s.send.retransmit_count); + try testing.expect(!s.send.hasUnackedData()); + + // Queue disposal and drain — stream fully removed, no leaks + const sid = s.stream_id; + sm.queueDisposal(sid); + sm.drainDisposalQueue(); + try testing.expect(sm.streams.get(sid) == null); +}