[HIGH] Bound peer-created stream state

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,24 +410,24 @@ 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
                 # parent is a no-op here, and a phantom stream would hold
                 # 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

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

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

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)

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