Improve concurrent streaming error handling and configuration

This commit addresses several critical issues and improvements for concurrent
component streaming in React on Rails Pro:

1. CRITICAL: Client Disconnect Handling
   - Added IOError and Errno::EPIPE exception handling in producer tasks
   - Added stream.closed? check before expensive operations
   - Added exception handling in writer task to stop gracefully on disconnect
   - Prevents wasted resources when clients disconnect mid-stream

2. Configuration Validation Improvements
   - Replaced attr_accessor with custom setter for concurrent_component_streaming_buffer_size
   - Added validation at assignment time, not just initialization
   - Added comprehensive documentation about memory implications
   - Documented that buffer size * chunk size determines max memory usage

3. Documentation Enhancements
   - Added detailed comments explaining producer-consumer pattern
   - Documented ordering guarantees for concurrent streaming
   - Clarified that chunks from same component maintain order
   - Clarified that chunks from different components may interleave
   - Added memory management documentation to drain_streams_concurrently

4. Testing
   - Added test for client disconnect scenario
   - Verifies producer stops when client disconnects
   - Ensures no further processing after IOError

Changes:
- react_on_rails_pro/lib/react_on_rails_pro/concerns/stream.rb
- react_on_rails_pro/lib/react_on_rails_pro/configuration.rb
- react_on_rails_pro/spec/dummy/spec/helpers/react_on_rails_pro_helper_spec.rb

Note: Did not implement sleep-based test synchronization improvements as the
existing approach is acceptable for test purposes and more complex synchronization
mechanisms would add unnecessary complexity to the test suite.

Author: justin808

react_on_rails_pro/lib/react_on_rails_pro/concerns/stream.rb

@@ -47,6 +47,20 @@ def stream_view_containing_react_components(template:, close_stream_at_end: true
 
     private
 
+    # Drains all streaming fibers concurrently using a producer-consumer pattern.
+    #
+    # Producer tasks: Each fiber drains its stream and enqueues chunks to a shared queue.
+    # Consumer task: Single writer dequeues chunks and writes them to the response stream.
+    #
+    # Ordering guarantees:
+    # - Chunks from the same component maintain their order
+    # - Chunks from different components may interleave based on production timing
+    # - The first component to produce a chunk will have it written first
+    #
+    # Memory management:
+    # - Uses a limited queue (configured via concurrent_component_streaming_buffer_size)
+    # - Producers block when the queue is full, providing backpressure
+    # - This prevents unbounded memory growth from fast producers
     def drain_streams_concurrently
       require "async"
       require "async/limited_queue"
@@ -58,7 +72,9 @@ def drain_streams_concurrently
         buffer_size = ReactOnRailsPro.configuration.concurrent_component_streaming_buffer_size
         queue = Async::LimitedQueue.new(buffer_size)
 
+        # Consumer task: Single writer dequeues and writes to response stream
         writer = build_writer_task(parent: parent, queue: queue)
+        # Producer tasks: Each fiber drains its stream and enqueues chunks
         tasks = build_producer_tasks(parent: parent, queue: queue)
 
         # This structure ensures that even if a producer task fails, we always
@@ -78,12 +94,18 @@ def build_producer_tasks(parent:, queue:)
       @rorp_rendering_fibers.each_with_index.map do |fiber, idx|
         parent.async do
           loop do
+            # Check if client disconnected before expensive operations
+            break if response.stream.closed?
+
             chunk = fiber.resume
             break unless chunk
 
             # Will be blocked if the queue is full until a chunk is dequeued
             queue.enqueue([idx, chunk])
           end
+        rescue IOError, Errno::EPIPE
+          # Client disconnected - stop producing
+          break
         end
       end
     end
@@ -97,6 +119,9 @@ def build_writer_task(parent:, queue:)
           _idx_from_queue, item = pair
           response.stream.write(item)
         end
+      rescue IOError, Errno::EPIPE
+        # Client disconnected - stop writing
+        nil
       end
     end
   end

react_on_rails_pro/lib/react_on_rails_pro/configuration.rb

@@ -70,7 +70,31 @@ class Configuration # rubocop:disable Metrics/ClassLength
                   :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, :rsc_bundle_js_file, :react_client_manifest_file,
-                  :react_server_client_manifest_file, :concurrent_component_streaming_buffer_size
+                  :react_server_client_manifest_file
+
+    attr_reader :concurrent_component_streaming_buffer_size
+
+    # Sets the buffer size for concurrent component streaming.
+    #
+    # This value controls how many chunks can be buffered in memory during
+    # concurrent streaming operations. When producers generate chunks faster
+    # than they can be written to the client, this buffer prevents unbounded
+    # memory growth by blocking producers when the buffer is full.
+    #
+    # Memory implications:
+    # - Buffer size of 64 (default) with 1KB chunks = ~64KB max memory
+    # - Buffer size of 64 with 1MB chunks = ~64MB max memory
+    # - Consider your typical chunk size when configuring this value
+    #
+    # @param value [Integer] A positive integer specifying the buffer size
+    # @raise [ReactOnRailsPro::Error] if value is not a positive integer
+    def concurrent_component_streaming_buffer_size=(value)
+      unless value.is_a?(Integer) && value.positive?
+        raise ReactOnRailsPro::Error,
+              "config.concurrent_component_streaming_buffer_size must be a positive integer"
+      end
+      @concurrent_component_streaming_buffer_size = value
+    end
 
     def initialize(renderer_url: nil, renderer_password: nil, server_renderer: nil, # rubocop:disable Metrics/AbcSize
                    renderer_use_fallback_exec_js: nil, prerender_caching: nil,

react_on_rails_pro/spec/dummy/spec/helpers/react_on_rails_pro_helper_spec.rb

@@ -512,6 +512,57 @@ def execute_stream_view_containing_react_components
           expect(chunk).not_to include("<h1>Header Rendered In View</h1>")
         end
       end
+
+      it "stops producing when client disconnects" do
+        queue = Async::Queue.new
+        mock_request_and_response(queue)
+
+        # Simulate client disconnect by making stream.write raise IOError
+        call_count = 0
+        allow(mocked_stream).to receive(:write) do |chunk|
+          call_count += 1
+          written_chunks << chunk
+          raise IOError, "client disconnected" if call_count == 2
+        end
+
+        # Configure render_to_string to call stream_react_component
+        allow(self).to receive(:render_to_string) do
+          render_result = stream_react_component(component_name, props: props, **component_options)
+          <<-HTML
+            <div>
+              <h1>Header Rendered In View</h1>
+              #{render_result}
+            </div>
+          HTML
+        end
+
+        Sync do |parent|
+          parent.async do
+            stream_view_containing_react_components(template: template_path)
+          rescue IOError
+            # Expected - client disconnected
+          end
+
+          # Enqueue first chunk - should be written successfully
+          queue.enqueue(chunks[0])
+          sleep 0.05
+
+          # Enqueue second chunk - should trigger disconnect
+          queue.enqueue(chunks[1])
+          sleep 0.05
+
+          # Enqueue third chunk - should not be written (producer stopped)
+          queue.enqueue(chunks[2])
+          sleep 0.05
+
+          queue.close
+          sleep 0.05
+        end
+
+        # Should only have written the first chunk before disconnect
+        expect(written_chunks.count).to eq(1)
+        expect(mocked_stream).to have_received(:write).at_most(2).times
+      end
     end
 
     describe "#cached_stream_react_component", :caching do # rubocop:disable RSpec/MultipleMemoizedHelpers