Discard late HEADERS for locally reset streams

lib/protocol/http2/connection.rb

@@ -510,6 +510,15 @@ def receive_headers(frame)
 
 				if stream = @streams[stream_id]
 					stream.receive_headers(frame)
+				elsif local_stream_id?(stream_id) and closed_stream_id?(stream_id)
+					# This can occur if we reset a stream while the remote peer was sending headers for it,
+					# e.g. a response which was already in flight when the request was cancelled.
+					# The header block must still be decoded in order to keep the HPACK decoder state
+					# synchronized with the remote peer's encoder, but the decoded headers are discarded
+					# (RFC 9113 §5.1).
+					decode_headers(frame.unpack)
+
+					return nil
 				else
 					if stream_id <= @remote_stream_id
 						raise ProtocolError, "Invalid stream id: #{stream_id} <= #{@remote_stream_id}!"

lib/protocol/http2/continuation_frame.rb

@@ -90,7 +90,7 @@ def pack(data, **options)
 
 					remainder = data.byteslice(maximum_size, data.bytesize-maximum_size)
 
-					@continuation = ContinuationFrame.new
+					@continuation = ContinuationFrame.new(@stream_id)
 					@continuation.pack(remainder, maximum_size: maximum_size)
 				else
 					set_flags(END_HEADERS)

releases.md

@@ -1,5 +1,10 @@
 # Releases
 
+## Unreleased
+
+  - Decode and discard `HEADERS` received for a locally-initiated stream which was already reset, e.g. a response in flight when the request was cancelled, rather than failing the connection with `ProtocolError`. This keeps the HPACK decoder state synchronized with the remote peer (RFC 9113 §5.1).
+  - `CONTINUATION` frames generated when packing a large header block now carry the stream ID of the frame they continue.
+
 ## v0.28.0
 
   - Treat `RST_STREAM(NO_ERROR)` as an orderly stream closure rather than constructing a `StreamError`.

test/protocol/http2/connection.rb

@@ -366,6 +366,124 @@ def before
 			end.to raise_exception(Protocol::HTTP2::ProtocolError, message: be =~ /Invalid stream id/)
 		end
 
+		with "response headers in flight when the stream is cancelled" do
+			# The custom header is added to the HPACK dynamic table when it is first encoded, so subsequent header blocks refer to it by index:
+			let(:response_headers) {[[":status", "200"], ["x-request-id", "7f0c9a1e"]]}
+
+			def before
+				super
+
+				stream.send_headers(request_headers, Protocol::HTTP2::END_STREAM)
+				server.read_frame
+			end
+
+			def cancel_stream
+				stream.send_reset_stream(Protocol::HTTP2::CANCEL)
+
+				expect(stream.state).to be == :closed
+				expect(client.streams).not.to have_keys(stream.id)
+
+				frame = server.read_frame
+				expect(frame).to be_a(Protocol::HTTP2::ResetStreamFrame)
+				expect(server.streams).not.to have_keys(stream.id)
+			end
+
+			def write_headers(stream_id, data, **options)
+				frame = Protocol::HTTP2::HeadersFrame.new(stream_id, Protocol::HTTP2::END_STREAM)
+				frame.pack(data, **options)
+
+				yield frame if block_given?
+
+				server.write_frame(frame)
+
+				return frame
+			end
+
+			def expect_synchronized_decoder
+				another_stream = client.create_stream
+				another_stream.send_headers(request_headers, Protocol::HTTP2::END_STREAM)
+				server.read_frame
+
+				server.streams[another_stream.id].send_headers(response_headers, Protocol::HTTP2::END_STREAM)
+
+				expect(another_stream).to receive(:process_headers) do |frame|
+					headers = super(frame)
+
+					expect(headers).to be == response_headers
+				end
+
+				frame = client.read_frame
+				expect(frame).to be_a(Protocol::HTTP2::HeadersFrame)
+
+				# Both fields are encoded as a single byte index, the custom header referring to the dynamic table entry introduced by the discarded header block:
+				expect(frame.unpack.bytesize).to be == 2
+
+				expect(another_stream.state).to be == :closed
+			end
+
+			it "discards the headers and keeps the connection usable" do
+				server.streams[stream.id].send_headers(response_headers, Protocol::HTTP2::END_STREAM)
+
+				cancel_stream
+
+				frame = client.read_frame
+				expect(frame).to be_a(Protocol::HTTP2::HeadersFrame)
+				expect(frame.stream_id).to be == stream.id
+
+				expect(client).not.to be(:closed?)
+				expect(client.streams).not.to have_keys(stream.id)
+
+				expect_synchronized_decoder
+			end
+
+			it "decodes the complete header block split across continuation frames" do
+				cancel_stream
+
+				frame = write_headers(stream.id, server.encode_headers(response_headers), maximum_size: 4)
+				expect(frame).to be(:continued?)
+
+				expect(client.read_frame).to be_a(Protocol::HTTP2::HeadersFrame)
+				expect(client).not.to be(:closed?)
+				expect(client.streams).not.to have_keys(stream.id)
+
+				expect_synchronized_decoder
+			end
+
+			it "rejects malformed header blocks" do
+				cancel_stream
+
+				write_headers(stream.id, "\xFF".b)
+
+				expect do
+					client.read_frame
+				end.to raise_exception(Protocol::HPACK::Error)
+
+				expect(client).to be(:closed?)
+			end
+
+			it "rejects invalid continuation sequences" do
+				cancel_stream
+
+				write_headers(stream.id, server.encode_headers(response_headers), maximum_size: 4) do |frame|
+					frame.continuation.stream_id = stream.id + 2
+				end
+
+				expect do
+					client.read_frame
+				end.to raise_exception(Protocol::HTTP2::ProtocolError, message: be =~ /Invalid stream id/)
+			end
+
+			it "rejects headers for idle local streams" do
+				write_headers(stream.id + 2, server.encode_headers(response_headers))
+
+				expect do
+					client.read_frame
+				end.to raise_exception(Protocol::HTTP2::ProtocolError, message: be =~ /Invalid stream id/)
+
+				expect(client).to be(:closed?)
+			end
+		end
+
 		it "client can handle graceful shutdown" do
 			stream.send_headers(request_headers, Protocol::HTTP2::END_STREAM)
 

test/protocol/http2/headers_frame.rb

@@ -66,6 +66,14 @@ def before
 			expect(frame.continuation.length).to be == 3
 		end
 
+		it "generates continuation frames for the same stream" do
+			frame = subject.new(1)
+			frame.pack "Hello World, Goodbye World", maximum_size: 8
+
+			expect(frame.continuation.stream_id).to be == 1
+			expect(frame.continuation.continuation.stream_id).to be == 1
+		end
+
 		it "can read and write continuation frames" do
 			frame.write(stream)
 			stream.rewind