diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index 450c0eed..6b63ebb8 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -87,10 +87,13 @@ def initialize(settings = {}) @decompressor = Header::Decompressor.new(settings) @active_stream_count = 0 + @promised_stream_count = 0 @last_stream_id = 0 @streams = {} @streams_recently_closed = {} + @idle_stream_priorities = {} @oldest_stream_recently_closed = nil + @last_promised_stream_id = 0 @pending_settings = [] @framer = Framer.new(@local_settings[:settings_max_frame_size]) @@ -316,11 +319,19 @@ def receive(data) verify_pseudo_headers(frame) verify_stream_order(stream_id) + priority = @idle_stream_priorities.delete(stream_id) + if frame[:flags].anybits?(PRIORITY) + priority = { + weight: frame[:weight], + dependency: frame[:dependency], + exclusive: frame[:exclusive] + } + end stream = activate_stream( id: stream_id, - weight: frame[:weight] || DEFAULT_WEIGHT, - dependency: frame[:dependency] || 0, - exclusive: frame[:exclusive] || false + weight: priority&.[](:weight) || DEFAULT_WEIGHT, + dependency: priority&.[](:dependency) || 0, + exclusive: priority&.[](:exclusive) || false ) emit(:stream, stream) end @@ -352,6 +363,8 @@ def receive(data) parent = @streams[stream_id] pid = frame[:promise_stream] + validate_push_promise(pid) + # if PUSH parent is recently closed, RST_STREAM the push if @streams_recently_closed[stream_id] send(type: :rst_stream, stream: pid, error: :refused_stream) @@ -377,6 +390,12 @@ def receive(data) _verify_pseudo_headers(frame, REQUEST_MANDATORY_HEADERS) verify_stream_order(pid) + + if @promised_stream_count >= @local_settings[:settings_max_concurrent_streams] + send(type: :rst_stream, stream: pid, error: :refused_stream) + next + end + stream = activate_stream(id: pid, parent: parent) emit(:promise, stream) stream << frame @@ -391,7 +410,7 @@ def receive(data) case frame_type # Priority signaling is deprecated (RFC 9113 Section 5.3.2), but # a PRIORITY frame may still name a stream id not opened yet: - # activate it, so the priority sticks if the stream opens. + # retain its priority without creating a stream until it opens. when :priority # After a GOAWAY, a PRIORITY for a stream already used and # closed must not resurrect it: reprioritizing a closed @@ -399,16 +418,16 @@ def receive(data) # the draining connection open. next if closed? && stream_id <= @last_stream_id - stream = activate_stream( - id: stream_id, + if !@idle_stream_priorities.key?(stream_id) && + (@idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams]) + next + end + + @idle_stream_priorities[stream_id] = { weight: frame[:weight] || DEFAULT_WEIGHT, dependency: frame[:dependency] || 0, exclusive: frame[:exclusive] || false - ) - - emit(:stream, stream) - stream << frame - + } # WINDOW_UPDATE can be sent by a peer that has sent a frame # bearing the END_STREAM flag. This means that a receiver could # receive a WINDOW_UPDATE frame on a "half-closed (remote)" or @@ -577,6 +596,16 @@ def ping_management(frame) end end + def validate_push_promise(stream_id) + connection_error(:protocol_error, msg: "clients cannot send PUSH_PROMISE") if @remote_role == :client + connection_error(:protocol_error, msg: "push promises are disabled") if @local_settings[:settings_enable_push].zero? + unless stream_id.positive? && stream_id.even? && stream_id > @last_promised_stream_id + connection_error(:protocol_error, msg: "invalid promised stream ID") + end + + @last_promised_stream_id = stream_id + end + # Validate settings parameters. See sepc Section 6.5.2. # # @param role [Symbol] The sender's role: :client or :server @@ -805,9 +834,11 @@ def activate_stream(id:, max_concurrent_streams: @local_settings[:settings_max_c raise StreamLimitExceeded if @active_stream_count >= max_concurrent_streams stream = Stream.new(connection: self, id: id, **args) + @promised_stream_count += 1 if stream.parent stream.once(:close) do @streams.delete(id) + @promised_stream_count -= 1 if stream.parent # A graceful GOAWAY leaves the connection :closing until the # tracked streams complete (Section 6.8); the last one to close diff --git a/sig/connection.rbs b/sig/connection.rbs index 38ca5fb8..74a7bee4 100644 --- a/sig/connection.rbs +++ b/sig/connection.rbs @@ -29,11 +29,14 @@ module HTTP2 @stream_id: Integer @active_stream_count: Integer + @promised_stream_count: Integer @last_stream_id: Integer @streams: Hash[Integer, Stream] @streams_recently_closed: Hash[Integer, Numeric] + @idle_stream_priorities: Hash[Integer, Hash[Symbol, Integer | bool]] @oldest_stream_recently_closed: Numeric + @last_promised_stream_id: Integer @framer: Framer @@ -91,6 +94,8 @@ module HTTP2 def ping_management: (frame) -> void + def validate_push_promise: (Integer stream_id) -> void + def validate_settings: (role_type, settings_enum) -> void def connection_settings: (settings_frame) -> void diff --git a/spec/connection_spec.rb b/spec/connection_spec.rb index 7c358b18..da9e711e 100644 --- a/spec/connection_spec.rb +++ b/spec/connection_spec.rb @@ -8,6 +8,17 @@ let(:conn) { Client.new } let(:f) { Framer.new } + it "tracks promised streams until their close callback without scanning streams" do + parent = conn.new_stream + promised = conn.__send__(:activate_stream, id: 2, parent: parent) + expect(conn.instance_variable_get(:@promised_stream_count)).to eq 1 + + promised.emit(:close) + expect(conn.instance_variable_get(:@promised_stream_count)).to eq 0 + promised.emit(:close) + expect(conn.instance_variable_get(:@promised_stream_count)).to eq 0 + end + context "Headers pre/post processing" do let(:conn) do client = Client.new diff --git a/spec/server_spec.rb b/spec/server_spec.rb index 86177149..c7e75760 100644 --- a/spec/server_spec.rb +++ b/spec/server_spec.rb @@ -153,6 +153,19 @@ expect(stream.weight).to eq 20 end + it "should preserve updated PRIORITY values when an idle stream opens" do + srv << CONNECTION_PREFACE_MAGIC + srv << f.generate(settings_frame) + + streams = [] + srv.on(:stream) { |stream| streams << stream } + srv << f.generate(priority_frame.merge(stream: 1, weight: 20)) + srv << f.generate(priority_frame.merge(stream: 1, weight: 30)) + srv << f.generate(headers_frame) + + expect(streams.last.weight).to eq 30 + end + it "should process connection management frames after GOAWAY" do srv << CONNECTION_PREFACE_MAGIC srv << f.generate(settings_frame) diff --git a/spec/shared_examples/connection.rb b/spec/shared_examples/connection.rb index d4d0bcd7..0d6f9308 100644 --- a/spec/shared_examples/connection.rb +++ b/spec/shared_examples/connection.rb @@ -317,14 +317,18 @@ expect(conn.active_stream_count).to eq(limit + 50) end - it "should initialize idle stream on PRIORITY frame" do + it "should retain idle priority without initializing a stream" do conn << f.generate(settings_frame) stream = nil conn.on(:stream) { |s| stream = s } conn << f.generate(priority_frame) - expect(stream.state).to eq :idle + expect(stream).to be_nil + expect(conn.instance_variable_get(:@streams)).to be_empty + expect(conn.instance_variable_get(:@idle_stream_priorities)[1]).to eq( + weight: 20, dependency: 0, exclusive: false + ) end end