diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index 450c0ee..2610cd8 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -90,7 +90,9 @@ def initialize(settings = {}) @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 +318,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.fetch(:weight, DEFAULT_WEIGHT), + dependency: priority.fetch(:dependency, 0), + exclusive: priority.fetch(:exclusive, false) ) emit(:stream, stream) end @@ -352,6 +362,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 +389,13 @@ def receive(data) _verify_pseudo_headers(frame, REQUEST_MANDATORY_HEADERS) verify_stream_order(pid) + + promised_stream_count = @streams.each_value.count(&:parent) + 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 @@ -399,12 +418,17 @@ def receive(data) # the draining connection open. next if closed? && stream_id <= @last_stream_id - stream = activate_stream( - id: stream_id, + unless @idle_stream_priorities.key?(stream_id) + next if @idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams] + end + + priority = { weight: frame[:weight] || DEFAULT_WEIGHT, dependency: frame[:dependency] || 0, exclusive: frame[:exclusive] || false - ) + } + @idle_stream_priorities[stream_id] = priority + stream = Stream.new(connection: self, id: stream_id, **priority) emit(:stream, stream) stream << frame @@ -577,6 +601,18 @@ def ping_management(frame) end end + def validate_push_promise(stream_id) + connection_error(:protocol_error, msg: "clients cannot send PUSH_PROMISE") if @local_role == :server + if @local_settings[:settings_enable_push].zero? + connection_error(:protocol_error, msg: "received PUSH_PROMISE while push is disabled") + end + 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 diff --git a/sig/connection.rbs b/sig/connection.rbs index 38ca5fb..91de8bf 100644 --- a/sig/connection.rbs +++ b/sig/connection.rbs @@ -33,7 +33,9 @@ module HTTP2 @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 +93,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/server_spec.rb b/spec/server_spec.rb index 8617714..c7e7576 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)