remove the concurrent_stream_drain config flag and always stream components concurrently

Author: ihabadham

lib/react_on_rails_pro/concerns/stream.rb

@@ -39,11 +39,7 @@ def stream_view_containing_react_components(template:, close_stream_at_end: true
       response.stream.write(template_string)
 
       begin
-        if ReactOnRailsPro.configuration.concurrent_stream_drain
-          drain_streams_concurrently
-        else
-          drain_streams_sequentially
-        end
+        drain_streams_concurrently
       ensure
         response.stream.close if close_stream_at_end
       end
@@ -122,22 +118,6 @@ def build_writer_task(parent:, queue:, semaphore:)
       end
     end
 
-    def drain_streams_sequentially
-      @rorp_rendering_fibers.each_with_index do |fiber, idx|
-        loop do
-          begin
-            chunk = fiber.resume
-          rescue FiberError
-            break
-          end
-          break unless chunk
-
-          log_stream_write(mode: :sequential, idx: idx, bytesize: safe_bytesize(chunk))
-          response.stream.write(chunk)
-        end
-      end
-    end
-
     def log_stream_write(mode:, idx:, bytesize:)
       return unless ReactOnRailsPro.configuration.tracing
 

lib/react_on_rails_pro/configuration.rb

@@ -53,7 +53,6 @@ class Configuration # rubocop:disable Metrics/ClassLength
     DEFAULT_RAISE_NON_SHELL_SERVER_RENDERING_ERRORS = false
     DEFAULT_ENABLE_RSC_SUPPORT = false
     DEFAULT_RSC_PAYLOAD_GENERATION_URL_PATH = "rsc_payload/"
-    DEFAULT_CONCURRENT_STREAM_DRAIN = false
     DEFAULT_CONCURRENT_STREAM_QUEUE_CAPACITY = 64
 
     attr_accessor :renderer_url, :renderer_password, :tracing,
@@ -63,7 +62,7 @@ class Configuration # rubocop:disable Metrics/ClassLength
                   :remote_bundle_cache_adapter, :ssr_pre_hook_js, :assets_to_copy,
                   :renderer_request_retry_limit, :throw_js_errors, :ssr_timeout,
                   :profile_server_rendering_js_code, :raise_non_shell_server_rendering_errors, :enable_rsc_support,
-                  :rsc_payload_generation_url_path, :concurrent_stream_drain, :concurrent_stream_queue_capacity
+                  :rsc_payload_generation_url_path, :concurrent_stream_queue_capacity
 
     def initialize(renderer_url: nil, renderer_password: nil, server_renderer: nil, # rubocop:disable Metrics/AbcSize
                    renderer_use_fallback_exec_js: nil, prerender_caching: nil,
@@ -74,7 +73,6 @@ def initialize(renderer_url: nil, renderer_password: nil, server_renderer: nil,
                    renderer_request_retry_limit: nil, throw_js_errors: nil, ssr_timeout: nil,
                    profile_server_rendering_js_code: nil, raise_non_shell_server_rendering_errors: nil,
                    enable_rsc_support: nil, rsc_payload_generation_url_path: nil,
-                   concurrent_stream_drain: DEFAULT_CONCURRENT_STREAM_DRAIN,
                    concurrent_stream_queue_capacity: DEFAULT_CONCURRENT_STREAM_QUEUE_CAPACITY)
       self.renderer_url = renderer_url
       self.renderer_password = renderer_password
@@ -98,7 +96,6 @@ def initialize(renderer_url: nil, renderer_password: nil, server_renderer: nil,
       self.raise_non_shell_server_rendering_errors = raise_non_shell_server_rendering_errors
       self.enable_rsc_support = enable_rsc_support
       self.rsc_payload_generation_url_path = rsc_payload_generation_url_path
-      self.concurrent_stream_drain = concurrent_stream_drain
       self.concurrent_stream_queue_capacity = concurrent_stream_queue_capacity
     end
 

spec/react_on_rails_pro/stream_spec.rb

@@ -372,14 +372,6 @@ def render_to_string(**_opts)
   end
 
   describe "Component streaming concurrency" do
-    def with_concurrent_stream_drain(enabled)
-      original = ReactOnRailsPro.configuration.concurrent_stream_drain
-      ReactOnRailsPro.configuration.concurrent_stream_drain = enabled
-      yield
-    ensure
-      ReactOnRailsPro.configuration.concurrent_stream_drain = original
-    end
-
     def run_stream(controller, template: "ignored")
       Sync do |parent|
         parent.async { controller.stream_view_containing_react_components(template: template) }
@@ -401,75 +393,42 @@ def setup_stream_test(component_count: 2)
       [component_queues, controller, mocked_stream]
     end
 
-    it "processes components concurrently vs sequentially" do
-      seq_queues, seq_controller, seq_stream = setup_stream_test
-
-      with_concurrent_stream_drain(false) do
-        run_stream(seq_controller) do |_parent|
-          seq_queues[0].enqueue("A1")
-          seq_queues[0].enqueue("A2")
-          seq_queues[0].close
-
-          seq_queues[1].enqueue("B1")
-          seq_queues[1].enqueue("B2")
-          seq_queues[1].close
-
-          sleep 0.1
-        end
-      end
-
-      # Verify sequential behavior: all A chunks processed before B chunks start
-      expect(seq_stream).to have_received(:write).with("TEMPLATE")
-      expect(seq_stream).to have_received(:write).with("A1") 
-      expect(seq_stream).to have_received(:write).with("A2")
-      expect(seq_stream).to have_received(:write).with("B1")
-      expect(seq_stream).to have_received(:write).with("B2")
-
-      conc_queues, conc_controller, conc_stream = setup_stream_test
+    it "streams components concurrently" do
+      queues, controller, stream = setup_stream_test
 
-      with_concurrent_stream_drain(true) do
-        run_stream(conc_controller) do |_parent|
-          conc_queues[1].enqueue("B1")
-          sleep 0.05
-          expect(conc_stream).to have_received(:write).with("B1")
+      run_stream(controller) do |_parent|
+        queues[1].enqueue("B1")
+        sleep 0.05
+        expect(stream).to have_received(:write).with("B1")
 
-          conc_queues[0].enqueue("A1")
-          sleep 0.05
-          expect(conc_stream).to have_received(:write).with("A1")
+        queues[0].enqueue("A1")
+        sleep 0.05
+        expect(stream).to have_received(:write).with("A1")
 
-          conc_queues[1].enqueue("B2")
-          conc_queues[1].close
-          sleep 0.05
+        queues[1].enqueue("B2")
+        queues[1].close
+        sleep 0.05
 
-          conc_queues[0].enqueue("A2")
-          conc_queues[0].close
-          sleep 0.1
-        end
+        queues[0].enqueue("A2")
+        queues[0].close
+        sleep 0.1
       end
-
-    # Verify concurrent behavior: components can process interleaved
-    expect(conc_stream).to have_received(:write).with("B1")
-    expect(conc_stream).to have_received(:write).with("A1") 
-    expect(conc_stream).to have_received(:write).with("B2")
-    expect(conc_stream).to have_received(:write).with("A2")
     end
 
     it "maintains per-component ordering" do
       queues, controller, stream = setup_stream_test
 
-      with_concurrent_stream_drain(true) do
-        run_stream(controller) do |_parent|
-          queues[0].enqueue("X1")
-          queues[0].enqueue("X2")
-          queues[0].enqueue("X3")
-          queues[0].close
+      run_stream(controller) do |_parent|
+        queues[0].enqueue("X1")
+        queues[0].enqueue("X2")
+        queues[0].enqueue("X3")
+        queues[0].close
 
-          queues[1].enqueue("Y1")
-          queues[1].enqueue("Y2")
-          queues[1].close
+        queues[1].enqueue("Y1")
+        queues[1].enqueue("Y2")
+        queues[1].close
 
-          sleep 0.2
-        end
+        sleep 0.2
       end
 
       # Verify all chunks were written
@@ -483,10 +442,8 @@ def setup_stream_test(component_count: 2)
     it "handles empty component list" do
       queues, controller, stream = setup_stream_test(component_count: 0)
 
-      with_concurrent_stream_drain(true) do
-        run_stream(controller) do |_parent|
-          sleep 0.1
-        end
+      run_stream(controller) do |_parent|
+        sleep 0.1
       end
 
       expect(stream).to have_received(:write).with("TEMPLATE")
@@ -496,14 +453,12 @@ def setup_stream_test(component_count: 2)
     it "handles single component" do
       queues, controller, stream = setup_stream_test(component_count: 1)
 
-      with_concurrent_stream_drain(true) do
-        run_stream(controller) do |_parent|
-          queues[0].enqueue("Single1")
-          queues[0].enqueue("Single2")
-          queues[0].close
+      run_stream(controller) do |_parent|
+        queues[0].enqueue("Single1")
+        queues[0].enqueue("Single2")
+        queues[0].close
 
-          sleep 0.1
-        end
+        sleep 0.1
       end
 
       expect(stream).to have_received(:write).with("Single1")
@@ -519,13 +474,11 @@ def setup_stream_test(component_count: 2)
         sleep 0.05
       end
 
-      with_concurrent_stream_drain(true) do
-        run_stream(controller) do |_parent|
-          5.times { |i| queues[0].enqueue("Chunk#{i}") }
-          queues[0].close
+      run_stream(controller) do |_parent|
+        5.times { |i| queues[0].enqueue("Chunk#{i}") }
+        queues[0].close
 
-          sleep 1
-        end
+        sleep 1
       end
 
       expect(write_timestamps.length).to be >= 2