diff --git a/build.zig.zon b/build.zig.zon index 3cb932f..3a6e2e8 100644 --- a/build.zig.zon +++ b/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 = .{ diff --git a/src/frames/transport.zig b/src/frames/transport.zig index 241c016..86f8986 100644 --- a/src/frames/transport.zig +++ b/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 { diff --git a/src/transport/io.zig b/src/transport/io.zig index 7bcd609..315564d 100644 --- a/src/transport/io.zig +++ b/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); +} diff --git a/src/transport/raw_app_stream.zig b/src/transport/raw_app_stream.zig index 559fb6f..6457fc0 100644 --- a/src/transport/raw_app_stream.zig +++ b/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