From d280ca09d80454a77bef914d42b16b142d31a7b4 Mon Sep 17 00:00:00 2001 From: Oskar Eichler Date: Fri, 4 Sep 2026 19:57:55 +0700 Subject: [PATCH 1/5] Bound peer-created stream state --- lib/http/2/connection.rb | 30 +++++++++++++++++++++++++++++- sig/connection.rbs | 4 ++++ 2 files changed, 33 insertions(+), 1 deletion(-) diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index 450c0eed..bd77d99a 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_priority_stream_ids = {} @oldest_stream_recently_closed = nil + @last_promised_stream_id = 0 @pending_settings = [] @framer = Framer.new(@local_settings[:settings_max_frame_size]) @@ -352,6 +354,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 +381,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,7 +410,12 @@ def receive(data) # the draining connection open. next if closed? && stream_id <= @last_stream_id - stream = activate_stream( + next if @idle_priority_stream_ids.key?(stream_id) + next if @idle_priority_stream_ids.size >= @local_settings[:settings_max_concurrent_streams] + + @idle_priority_stream_ids[stream_id] = true + stream = Stream.new( + connection: self, id: stream_id, weight: frame[:weight] || DEFAULT_WEIGHT, dependency: frame[:dependency] || 0, @@ -577,6 +593,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 38ca5fb8..6982c69f 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_priority_stream_ids: Hash[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 From 2b4f59998fe50b99eaf9710acd36d72c48565ca9 Mon Sep 17 00:00:00 2001 From: Oskar Eichler <62393985+OskarEichler@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:05:45 +0700 Subject: [PATCH 2/5] Preserve bounded idle stream priorities --- lib/http/2/connection.rb | 23 ++++++++++++----------- sig/connection.rbs | 2 +- spec/server_spec.rb | 13 +++++++++++++ 3 files changed, 26 insertions(+), 12 deletions(-) diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index bd77d99a..bf9cabbd 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -90,7 +90,7 @@ def initialize(settings = {}) @last_stream_id = 0 @streams = {} @streams_recently_closed = {} - @idle_priority_stream_ids = {} + @idle_stream_priorities = {} @oldest_stream_recently_closed = nil @last_promised_stream_id = 0 @pending_settings = [] @@ -318,11 +318,12 @@ def receive(data) verify_pseudo_headers(frame) verify_stream_order(stream_id) + priority = @idle_stream_priorities.delete(stream_id) || {} stream = activate_stream( id: stream_id, - weight: frame[:weight] || DEFAULT_WEIGHT, - dependency: frame[:dependency] || 0, - exclusive: frame[:exclusive] || false + weight: frame.fetch(:weight, priority.fetch(:weight, DEFAULT_WEIGHT)), + dependency: frame.fetch(:dependency, priority.fetch(:dependency, 0)), + exclusive: frame.fetch(:exclusive, priority.fetch(:exclusive, false)) ) emit(:stream, stream) end @@ -410,17 +411,17 @@ def receive(data) # the draining connection open. next if closed? && stream_id <= @last_stream_id - next if @idle_priority_stream_ids.key?(stream_id) - next if @idle_priority_stream_ids.size >= @local_settings[:settings_max_concurrent_streams] + unless @idle_stream_priorities.key?(stream_id) + next if @idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams] + end - @idle_priority_stream_ids[stream_id] = true - stream = Stream.new( - connection: self, - id: stream_id, + 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 diff --git a/sig/connection.rbs b/sig/connection.rbs index 6982c69f..91de8bf8 100644 --- a/sig/connection.rbs +++ b/sig/connection.rbs @@ -33,7 +33,7 @@ module HTTP2 @streams: Hash[Integer, Stream] @streams_recently_closed: Hash[Integer, Numeric] - @idle_priority_stream_ids: Hash[Integer, bool] + @idle_stream_priorities: Hash[Integer, Hash[Symbol, Integer | bool]] @oldest_stream_recently_closed: Numeric @last_promised_stream_id: Integer 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) From 7c4c97085a035b0a3126d60ea02dd735090ddf34 Mon Sep 17 00:00:00 2001 From: Oskar Eichler <62393985+OskarEichler@users.noreply.github.com> Date: Fri, 4 Sep 2026 23:15:44 +0700 Subject: [PATCH 3/5] Honor HEADERS priority overrides --- lib/http/2/connection.rb | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index bf9cabbd..2610cd8f 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -319,11 +319,18 @@ def receive(data) 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.fetch(:weight, priority.fetch(:weight, DEFAULT_WEIGHT)), - dependency: frame.fetch(:dependency, priority.fetch(:dependency, 0)), - exclusive: frame.fetch(:exclusive, priority.fetch(:exclusive, false)) + weight: priority.fetch(:weight, DEFAULT_WEIGHT), + dependency: priority.fetch(:dependency, 0), + exclusive: priority.fetch(:exclusive, false) ) emit(:stream, stream) end From 1a011bf1b1b1e0c765b2a0291527f01527a6279f Mon Sep 17 00:00:00 2001 From: Oskar Eichler Date: Sun, 27 Sep 2026 12:16:28 +0700 Subject: [PATCH 4/5] Track promised streams without repeated scans --- lib/http/2/connection.rb | 25 +++++++++++++------------ sig/connection.rbs | 1 + spec/connection_spec.rb | 11 +++++++++++ 3 files changed, 25 insertions(+), 12 deletions(-) diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index 2610cd8f..012cce89 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -87,6 +87,7 @@ def initialize(settings = {}) @decompressor = Header::Decompressor.new(settings) @active_stream_count = 0 + @promised_stream_count = 0 @last_stream_id = 0 @streams = {} @streams_recently_closed = {} @@ -318,7 +319,7 @@ def receive(data) verify_pseudo_headers(frame) verify_stream_order(stream_id) - priority = @idle_stream_priorities.delete(stream_id) || {} + priority = @idle_stream_priorities.delete(stream_id) if frame[:flags].anybits?(PRIORITY) priority = { weight: frame[:weight], @@ -328,9 +329,9 @@ def receive(data) end stream = activate_stream( id: stream_id, - weight: priority.fetch(:weight, DEFAULT_WEIGHT), - dependency: priority.fetch(:dependency, 0), - exclusive: priority.fetch(:exclusive, false) + weight: priority&.[](:weight) || DEFAULT_WEIGHT, + dependency: priority&.[](:dependency) || 0, + exclusive: priority&.[](:exclusive) || false ) emit(:stream, stream) end @@ -390,8 +391,7 @@ 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] + if @promised_stream_count >= @local_settings[:settings_max_concurrent_streams] send(type: :rst_stream, stream: pid, error: :refused_stream) next end @@ -418,8 +418,9 @@ def receive(data) # the draining connection open. next if closed? && stream_id <= @last_stream_id - unless @idle_stream_priorities.key?(stream_id) - next if @idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams] + if !@idle_stream_priorities.key?(stream_id) && + (@idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams]) + next end priority = { @@ -602,10 +603,8 @@ def ping_management(frame) 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 + 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 @@ -841,9 +840,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 91de8bf8..74a7bee4 100644 --- a/sig/connection.rbs +++ b/sig/connection.rbs @@ -29,6 +29,7 @@ module HTTP2 @stream_id: Integer @active_stream_count: Integer + @promised_stream_count: Integer @last_stream_id: Integer @streams: Hash[Integer, Stream] 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 From 82e298ca72af13431618bc092227ea78c187f023 Mon Sep 17 00:00:00 2001 From: Oskar Eichler Date: Fri, 2 Oct 2026 10:52:09 +0700 Subject: [PATCH 5/5] Retain idle priority metadata without creating streams --- lib/http/2/connection.rb | 10 ++-------- spec/shared_examples/connection.rb | 8 ++++++-- 2 files changed, 8 insertions(+), 10 deletions(-) diff --git a/lib/http/2/connection.rb b/lib/http/2/connection.rb index 012cce89..6b63ebb8 100644 --- a/lib/http/2/connection.rb +++ b/lib/http/2/connection.rb @@ -410,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 @@ -423,17 +423,11 @@ def receive(data) next end - priority = { + @idle_stream_priorities[stream_id] = { 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 - # 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 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