io: public raw-app stream RESET_STREAM / STOP_SENDING API

build.zig.zon

@@ -1,7 +1,7 @@
 .{
     .name = .zquic,
     .fingerprint = 0x947d823dd45c34db,
-    .version = "1.7.67",
+    .version = "1.7.68",
     .minimum_zig_version = "0.16.0",
     .dependencies = .{
         .zig_varint = .{

src/frames/transport.zig

@@ -44,6 +44,14 @@ pub const StopSending = struct {
         const code = try r.readVarint();
         return .{ .frame = .{ .stream_id = sid, .application_protocol_error_code = code }, .consumed = r.pos };
     }
+
+    pub fn serialize(self: StopSending, buf: []u8) (varint.EncodeError || varint.DecodeError)!usize {
+        var w = varint.Writer.init(buf);
+        try w.writeVarint(0x05);
+        try w.writeVarint(self.stream_id);
+        try w.writeVarint(self.application_protocol_error_code);
+        return w.pos;
+    }
 };
 
 pub const NewToken = struct {

src/transport/io.zig

@@ -5916,6 +5916,7 @@ pub const Server = struct {
                 // send- and receive-window slots are dead weight once reset.
                 conn.clearPeerStreamSendMax(r.frame.stream_id);
                 conn.clearStreamRecv(r.frame.stream_id);
+                markRawAppStreamReset(&conn.raw_app_streams, r.frame.stream_id, r.frame.application_protocol_error_code);
                 continue;
             }
             if (ft == 0x05) {
@@ -6144,6 +6145,27 @@ pub const Server = struct {
         self.send1Rtt(conn, frame_buf[0..frame_len], dst);
     }
 
+    /// Abort a raw-app send stream: send RESET_STREAM (RFC 9000 §19.4) with
+    /// `error_code` and free the local send slot. Embedder-facing counterpart
+    /// of `openRawAppStream` / `sendRawStreamData`.
+    pub fn resetRawAppStream(self: *Server, conn: *ConnState, stream_id: u64, error_code: u64) void {
+        self.sendResetStream(conn, stream_id, error_code, conn.peer);
+        conn.clearPeerStreamSendMax(stream_id);
+        _ = releaseRawAppStream(conn, stream_id, self.allocator);
+    }
+
+    /// Ask the peer to stop sending on `stream_id` (STOP_SENDING, RFC 9000
+    /// §19.5). The peer replies with RESET_STREAM.
+    pub fn stopSendingRawAppStream(self: *Server, conn: *ConnState, stream_id: u64, error_code: u64) void {
+        const frame = transport_frames.StopSending{
+            .stream_id = stream_id,
+            .application_protocol_error_code = error_code,
+        };
+        var frame_buf: [24]u8 = undefined;
+        const frame_len = frame.serialize(&frame_buf) catch return;
+        self.send1Rtt(conn, frame_buf[0..frame_len], conn.peer);
+    }
+
     /// Encrypt and send a 1-RTT packet, selecting AES or ChaCha20 per conn.
     fn send1Rtt(self: *Server, conn: *ConnState, payload: []const u8, dst: compat.Address) void {
         // RFC 9000 §10.2.3: do not send any frames while draining (only
@@ -8171,6 +8193,35 @@ pub fn rawAppRecvBuffer(conn: *ConnState, stream_id: u64) ?[]const u8 {
     return null;
 }
 
+/// Mark a raw-app stream as reset by the peer (called from the RESET_STREAM
+/// handlers on both roles — the server receives on `conn.raw_app_streams`, the
+/// client on `Client.raw_app_recv`). No-op if no active slot matches.
+fn markRawAppStreamReset(slots: []RawAppStreamSlot, stream_id: u64, error_code: u64) void {
+    for (slots) |*slot| {
+        if (slot.active and slot.stream_id == stream_id) {
+            slot.reset_received = true;
+            slot.reset_error_code = error_code;
+            return;
+        }
+    }
+}
+
+fn rawAppSlotsResetReceived(slots: []const RawAppStreamSlot, stream_id: u64) ?u64 {
+    for (slots) |*slot| {
+        if (slot.active and slot.stream_id == stream_id and slot.reset_received) {
+            return slot.reset_error_code;
+        }
+    }
+    return null;
+}
+
+/// If the peer reset `stream_id` (RESET_STREAM), returns its application error
+/// code; otherwise null. Mirrors the read side of Go transport
+/// `StreamResetError{Code}`. Server-side (streams received on the connection).
+pub fn rawAppStreamResetReceived(conn: *const ConnState, stream_id: u64) ?u64 {
+    return rawAppSlotsResetReceived(&conn.raw_app_streams, stream_id);
+}
+
 /// True when the peer has sent FIN on `stream_id` (one of the slots holds it
 /// and `fin_received` is set).  Embedders driving the libp2p
 /// per-message-stream pattern should call this after consuming the payload
@@ -9460,6 +9511,36 @@ pub const Client = struct {
         return null;
     }
 
+    /// Abort a raw-app send stream: send RESET_STREAM (RFC 9000 §19.4).
+    pub fn resetRawAppStream(self: *Client, stream_id: u64, error_code: u64) void {
+        const frame = transport_frames.ResetStream{
+            .stream_id = stream_id,
+            .application_protocol_error_code = error_code,
+            .final_size = 0,
+        };
+        var frame_buf: [32]u8 = undefined;
+        const frame_len = frame.serialize(&frame_buf) catch return;
+        _ = self.sendClient1Rtt(frame_buf[0..frame_len]);
+        self.conn.clearPeerStreamSendMax(stream_id);
+    }
+
+    /// Ask the peer to stop sending on `stream_id` (STOP_SENDING, RFC 9000 §19.5).
+    pub fn stopSendingRawAppStream(self: *Client, stream_id: u64, error_code: u64) void {
+        const frame = transport_frames.StopSending{
+            .stream_id = stream_id,
+            .application_protocol_error_code = error_code,
+        };
+        var frame_buf: [24]u8 = undefined;
+        const frame_len = frame.serialize(&frame_buf) catch return;
+        _ = self.sendClient1Rtt(frame_buf[0..frame_len]);
+    }
+
+    /// If the peer reset `stream_id` we were receiving, returns its app error
+    /// code; otherwise null (see `io.rawAppStreamResetReceived` for the server).
+    pub fn rawAppStreamResetReceived(self: *const Client, stream_id: u64) ?u64 {
+        return rawAppSlotsResetReceived(&self.raw_app_recv, stream_id);
+    }
+
     /// Mirror of the connection-level `rawAppStreamFinReceived`.
     pub fn rawAppStreamFinReceived(self: *const Client, stream_id: u64) bool {
         for (&self.raw_app_recv) |*slot| {
@@ -10863,6 +10944,7 @@ pub const Client = struct {
                 // per-stream send- and receive-window slots for this id.
                 self.conn.clearPeerStreamSendMax(r.frame.stream_id);
                 self.conn.clearStreamRecv(r.frame.stream_id);
+                markRawAppStreamReset(&self.raw_app_recv, r.frame.stream_id, r.frame.application_protocol_error_code);
                 continue;
             }
             if (ft == 0x05) {
@@ -14043,3 +14125,47 @@ test "resetForReconnect re-allocates the loss detector (resumption/0-RTT reconne
     try client.resetForReconnect(server_addr);
     try std.testing.expect(client.conn.ld.sent.len > 0);
 }
+
+test "resetRawAppStream: server resets a stream, client observes RESET_STREAM code (#40)" {
+    const allocator = std.testing.allocator;
+    const client = try allocator.create(Client);
+    defer allocator.destroy(client);
+
+    const lb = try rawSetupLoopback(allocator, client);
+    defer lb.server.deinit();
+    defer lb.client.deinit();
+
+    const conn = rawServerConnectedConn(lb.server).?;
+    const sid = try lb.server.openRawAppStream(conn);
+
+    // Send a chunk (no FIN) so the client allocates a recv slot for `sid`.
+    const payload = "hello-then-reset";
+    var drop = RawDrop{};
+    _ = lb.server.sendRawStreamData(conn, sid, 0, payload, false);
+    var delivered = false;
+    var i: usize = 0;
+    while (i < 400) : (i += 1) {
+        rawPumpOnce(lb.server, lb.client, lb.server_addr, &drop);
+        if (lb.client.rawAppRecvBuffer(sid)) |b| {
+            if (b.len == payload.len) {
+                delivered = true;
+                break;
+            }
+        }
+    }
+    try std.testing.expect(delivered);
+    try std.testing.expect(lb.client.rawAppStreamResetReceived(sid) == null); // not reset yet
+
+    // Reset the stream with application error code 42.
+    lb.server.resetRawAppStream(conn, sid, 42);
+    var seen_code: ?u64 = null;
+    i = 0;
+    while (i < 400) : (i += 1) {
+        rawPumpOnce(lb.server, lb.client, lb.server_addr, &drop);
+        if (lb.client.rawAppStreamResetReceived(sid)) |code| {
+            seen_code = code;
+            break;
+        }
+    }
+    try std.testing.expectEqual(@as(?u64, 42), seen_code);
+}

src/transport/raw_app_stream.zig

@@ -43,6 +43,10 @@ pub const RawAppStreamSlot = struct {
     /// Final stream size, recorded from the FIN frame (`offset + len`). Only
     /// meaningful once `fin_received` is true.
     fin_offset: u64 = 0,
+    /// True once the peer reset this stream with a RESET_STREAM frame
+    /// (RFC 9000 §19.4); `reset_error_code` carries the app error code.
+    reset_received: bool = false,
+    reset_error_code: u64 = 0,
 
     /// True only when the peer has FIN'd **and** all bytes up to the final
     /// size have been contiguously reassembled into `buf`. This is the signal