fix: Implement robust ACK handling and retransmission

- Update `build.zig.zon` to use `#` for `git+https` URLs.
- Refactor connection ACK tracking to manage `next_expected_packet`, `last_ack_sent`, `largest_peer_acked_packet`, `pending_ack`, and `received_packets`.
- Implement `recordReceivedPacket` to accurately track received packet numbers and return whether the packet is new.
- Add `scheduleAck`, `markAckSent`, and `getLargestPeerAckedPacket` to manage ACK state.
- Enhance `AckFrame` encoding and decoding to support multiple ACK ranges using QUIC's format.
- Improve `Stream::receiveData` to handle overlapping retransmits and deliver buffered chunks correctly.
- Add `bufferOutOfOrderChunk` to manage out-of-order stream data, replacing older chunks if new data overlaps.
- Modify `QuicTransport::processFrames` to determine if a frame is ACK-eliciting.
- Change `QuicTransport::sendPacket` to track retransmittable packets and enforce inflight capacity limits.
- Add `waitForInflightCapacity` to prevent unbounded inflight packets.
- Implement `buildAckFrameLayout` to construct ACK frames with proper range encoding based on received packets.
- Add `handleAckFrame`, `reapAckedPacketsForAckFrame`, and `packetAckedByAckFrame` to process received ACKs.
- Implement `fastRetransmitAckGaps` to retransmit lost packets indicated by ACK gaps.
- Implement `retransmitExpiredPackets` for timeout-based retransmissions using exponential backoff.
- Update `QuicTransport::flush` to call `retransmitExpiredPackets` before sending new data.
- Refactor `sendPacket` to `sendRawPacket` for sending individual UDP datagrams without encryption/framing.
- Add tests for new ACK and retransmission logic, including handling ACK-only payloads, fast retransmits, and continuous large writes.

Author: bresilla

build.zig.zon

@@ -3,7 +3,7 @@
     .version = "0.0.15",
     .dependencies = .{
         .libsafe = .{
-            .url = "git+https://github.com/libzig/libsafe?ref=0.0.7",
+            .url = "git+https://github.com/libzig/libsafe#0.0.7",
             .hash = "libsafe-0.0.7-ucZASCGdAwA9tuwbOrMk1dipWU9BfjIorGYIvasBxfaZ",
             // .path = "../libsafe",
         },

lib/connection.zig

@@ -18,7 +18,11 @@ pub const Connection = struct {
     /// Packet number tracking
     next_packet_number: u32,
     largest_received_packet: u32,
-    largest_acked_packet: u32,
+    next_expected_packet: u32,
+    last_ack_sent: ?u32,
+    largest_peer_acked_packet: ?u32,
+    pending_ack: bool,
+    received_packets: std.AutoHashMap(u32, void),
 
     /// Connection IDs
     local_conn_id: []const u8,
@@ -50,14 +54,21 @@ pub const Connection = struct {
 
         try streams.put(0, stream0);
 
+        var received_packets = std.AutoHashMap(u32, void).init(allocator);
+        errdefer received_packets.deinit();
+
         return Self{
             .allocator = allocator,
             .is_server = is_server,
             .streams = streams,
             .next_stream_id = initial_stream_id,
             .next_packet_number = 0,
             .largest_received_packet = 0,
-            .largest_acked_packet = 0,
+            .next_expected_packet = 0,
+            .last_ack_sent = null,
+            .largest_peer_acked_packet = null,
+            .pending_ack = false,
+            .received_packets = received_packets,
             .local_conn_id = local_conn_id,
             .remote_conn_id = remote_conn_id,
         };
@@ -71,6 +82,7 @@ pub const Connection = struct {
             self.allocator.destroy(stream.*);
         }
         self.streams.deinit();
+        self.received_packets.deinit();
     }
 
     /// Open a new bidirectional stream
@@ -146,17 +158,40 @@ pub const Connection = struct {
         return pn;
     }
 
-    /// Record received packet number
-    pub fn recordReceivedPacket(self: *Self, packet_number: u32) void {
+    /// Record received packet number.
+    ///
+    /// Returns true if the packet number is new.
+    pub fn recordReceivedPacket(self: *Self, packet_number: u32) !bool {
         if (packet_number > self.largest_received_packet) {
             self.largest_received_packet = packet_number;
         }
+
+        if (packet_number < self.next_expected_packet) {
+            return false;
+        }
+
+        const gop = try self.received_packets.getOrPut(packet_number);
+        if (gop.found_existing) {
+            return false;
+        }
+        gop.value_ptr.* = {};
+
+        while (self.received_packets.remove(self.next_expected_packet)) {
+            self.next_expected_packet += 1;
+        }
+
+        return true;
+    }
+
+    /// Schedule an ACK for transmission.
+    pub fn scheduleAck(self: *Self) void {
+        self.pending_ack = true;
     }
 
-    /// Record ACKed packet
-    pub fn recordAckedPacket(self: *Self, packet_number: u32) void {
-        if (packet_number > self.largest_acked_packet) {
-            self.largest_acked_packet = packet_number;
+    /// Record the largest packet number the peer has ACKed from us.
+    pub fn recordPeerAckedPacket(self: *Self, packet_number: u32) void {
+        if (self.largest_peer_acked_packet == null or packet_number > self.largest_peer_acked_packet.?) {
+            self.largest_peer_acked_packet = packet_number;
         }
     }
 
@@ -179,13 +214,22 @@ pub const Connection = struct {
 
     /// Check if connection needs to send ACK
     pub fn needsAck(self: *const Self) bool {
-        // Send ACK if we've received packets that haven't been acked
-        return self.largest_received_packet > self.largest_acked_packet;
+        return self.pending_ack and (self.next_expected_packet > 0 or self.received_packets.count() > 0);
     }
 
     /// Get packet number to acknowledge
     pub fn getAckNumber(self: *const Self) u32 {
-        return self.largest_received_packet;
+        std.debug.assert(self.next_expected_packet > 0);
+        return self.next_expected_packet - 1;
+    }
+
+    pub fn markAckSent(self: *Self) void {
+        self.last_ack_sent = if (self.next_expected_packet > 0) self.getAckNumber() else null;
+        self.pending_ack = false;
+    }
+
+    pub fn getLargestPeerAckedPacket(self: *const Self) ?u32 {
+        return self.largest_peer_acked_packet;
     }
 };
 
@@ -255,11 +299,83 @@ test "Connection - packet number tracking" {
     try testing.expectEqual(@as(u32, 2), conn.nextPacketNumber());
 
     // Record received packets
-    conn.recordReceivedPacket(5);
-    conn.recordReceivedPacket(10);
-    conn.recordReceivedPacket(7); // Out of order
+    try testing.expect(try conn.recordReceivedPacket(5));
+    try testing.expect(try conn.recordReceivedPacket(10));
+    try testing.expect(try conn.recordReceivedPacket(7)); // Out of order
 
     try testing.expectEqual(@as(u32, 10), conn.largest_received_packet);
+    try testing.expect(!conn.needsAck());
+}
+
+test "Connection - ACK scheduling is explicit" {
+    const testing = std.testing;
+    const allocator = testing.allocator;
+
+    var conn = try Connection.init(
+        allocator,
+        "local",
+        "remote",
+        false,
+    );
+    defer conn.deinit();
+
+    try testing.expect(try conn.recordReceivedPacket(0));
+    try testing.expect(!conn.needsAck());
+
+    conn.scheduleAck();
+    try testing.expect(conn.needsAck());
+}
+
+test "Connection - cumulative ACK tracking handles gaps" {
+    const testing = std.testing;
+    const allocator = testing.allocator;
+
+    var conn = try Connection.init(
+        allocator,
+        "local",
+        "remote",
+        false,
+    );
+    defer conn.deinit();
+
+    try testing.expect(try conn.recordReceivedPacket(0));
+    conn.scheduleAck();
+    try testing.expect(conn.needsAck());
+    try testing.expectEqual(@as(u32, 0), conn.getAckNumber());
+    conn.markAckSent();
+    try testing.expect(!conn.needsAck());
+
+    try testing.expect(try conn.recordReceivedPacket(2));
+    conn.scheduleAck();
+    try testing.expect(conn.needsAck());
+    try testing.expectEqual(@as(u32, 0), conn.getAckNumber());
+    conn.markAckSent();
+
+    try testing.expect(try conn.recordReceivedPacket(1));
+    conn.scheduleAck();
+    try testing.expect(conn.needsAck());
+    try testing.expectEqual(@as(u32, 2), conn.getAckNumber());
+}
+
+test "Connection - peer ACK state is separate from outbound ACK generation" {
+    const testing = std.testing;
+    const allocator = testing.allocator;
+
+    var conn = try Connection.init(
+        allocator,
+        "local",
+        "remote",
+        false,
+    );
+    defer conn.deinit();
+
+    conn.recordPeerAckedPacket(99);
+    try testing.expectEqual(@as(?u32, 99), conn.getLargestPeerAckedPacket());
+
+    try testing.expect(try conn.recordReceivedPacket(0));
+    conn.scheduleAck();
+    try testing.expect(conn.needsAck());
+    try testing.expectEqual(@as(u32, 0), conn.getAckNumber());
 }
 
 test "Connection - getOrCreateStream" {

lib/frame.zig

@@ -196,13 +196,23 @@ pub const MaxStreamDataFrame = struct {
     }
 };
 
-/// ACK Frame (simplified - just largest acknowledged)
+/// ACK Frame (simplified QUIC ACK ranges without ECN)
 pub const AckFrame = struct {
     largest_acknowledged: u64,
+    first_ack_range: u64 = 0,
+    ack_ranges: []const AckRange = &.{},
 
-    pub fn encode(self: AckFrame, buffer: []u8) !usize {
-        if (buffer.len < 10) return error.BufferTooSmall;
+    pub const AckRange = struct {
+        gap: u64,
+        ack_range_length: u64,
+    };
+
+    pub const DecodeResult = struct {
+        frame: AckFrame,
+        consumed: usize,
+    };
 
+    pub fn encode(self: AckFrame, buffer: []u8) !usize {
         var offset: usize = 0;
 
         // Type
@@ -215,22 +225,71 @@ pub const AckFrame = struct {
         // ACK Delay (simplified - set to 0)
         offset += encodeVarInt(0, buffer[offset..]);
 
-        // ACK Range Count (simplified - 0 ranges)
-        offset += encodeVarInt(0, buffer[offset..]);
+        // ACK Range Count
+        offset += encodeVarInt(self.ack_ranges.len, buffer[offset..]);
+
+        // First ACK Range
+        offset += encodeVarInt(self.first_ack_range, buffer[offset..]);
+
+        // ACK Ranges
+        for (self.ack_ranges) |ack_range| {
+            offset += encodeVarInt(ack_range.gap, buffer[offset..]);
+            offset += encodeVarInt(ack_range.ack_range_length, buffer[offset..]);
+        }
 
         return offset;
     }
 
-    pub fn decode(buffer: []const u8) !AckFrame {
+    pub fn decode(buffer: []const u8, ack_ranges_out: []AckRange) !DecodeResult {
         if (buffer.len < 2) return error.BufferTooSmall;
 
-        const offset: usize = 1; // Skip type byte
+        var offset: usize = 0;
+        if (buffer[offset] != @intFromEnum(FrameType.ack)) {
+            return error.InvalidFrameType;
+        }
+        offset += 1;
 
-        // Largest Acknowledged
         const largest_ack_result = try decodeVarInt(buffer[offset..]);
+        offset += largest_ack_result.bytes;
+
+        const ack_delay_result = try decodeVarInt(buffer[offset..]);
+        offset += ack_delay_result.bytes;
+
+        const ack_range_count_result = try decodeVarInt(buffer[offset..]);
+        offset += ack_range_count_result.bytes;
+        if (ack_range_count_result.value > ack_ranges_out.len) {
+            return error.BufferTooSmall;
+        }
 
-        return AckFrame{
-            .largest_acknowledged = largest_ack_result.value,
+        const first_ack_range_result = try decodeVarInt(buffer[offset..]);
+        offset += first_ack_range_result.bytes;
+        if (first_ack_range_result.value > largest_ack_result.value) {
+            return error.InvalidData;
+        }
+
+        var parsed_ranges: usize = 0;
+        var i: u64 = 0;
+        while (i < ack_range_count_result.value) : (i += 1) {
+            const gap_result = try decodeVarInt(buffer[offset..]);
+            offset += gap_result.bytes;
+
+            const range_len_result = try decodeVarInt(buffer[offset..]);
+            offset += range_len_result.bytes;
+
+            ack_ranges_out[parsed_ranges] = .{
+                .gap = gap_result.value,
+                .ack_range_length = range_len_result.value,
+            };
+            parsed_ranges += 1;
+        }
+
+        return .{
+            .frame = AckFrame{
+                .largest_acknowledged = largest_ack_result.value,
+                .first_ack_range = first_ack_range_result.value,
+                .ack_ranges = ack_ranges_out[0..parsed_ranges],
+            },
+            .consumed = offset,
         };
     }
 };
@@ -332,6 +391,34 @@ test "StreamFrame - encode and decode" {
     try testing.expectEqualSlices(u8, original.data, decoded.data);
 }
 
+test "AckFrame - encode and decode with ranges" {
+    const testing = std.testing;
+
+    const original = AckFrame{
+        .largest_acknowledged = 12,
+        .first_ack_range = 2,
+        .ack_ranges = &[_]AckFrame.AckRange{
+            .{ .gap = 0, .ack_range_length = 4 },
+            .{ .gap = 1, .ack_range_length = 1 },
+        },
+    };
+
+    var buffer: [128]u8 = undefined;
+    const encoded_len = try original.encode(&buffer);
+
+    var ack_ranges: [4]AckFrame.AckRange = undefined;
+    const decoded = try AckFrame.decode(buffer[0..encoded_len], &ack_ranges);
+
+    try testing.expectEqual(original.largest_acknowledged, decoded.frame.largest_acknowledged);
+    try testing.expectEqual(original.first_ack_range, decoded.frame.first_ack_range);
+    try testing.expectEqual(@as(usize, 2), decoded.frame.ack_ranges.len);
+    try testing.expectEqual(original.ack_ranges[0].gap, decoded.frame.ack_ranges[0].gap);
+    try testing.expectEqual(original.ack_ranges[0].ack_range_length, decoded.frame.ack_ranges[0].ack_range_length);
+    try testing.expectEqual(original.ack_ranges[1].gap, decoded.frame.ack_ranges[1].gap);
+    try testing.expectEqual(original.ack_ranges[1].ack_range_length, decoded.frame.ack_ranges[1].ack_range_length);
+    try testing.expectEqual(encoded_len, decoded.consumed);
+}
+
 test "varInt encoding" {
     const testing = std.testing;