fix: make the CDP send path thread-safe

lib/ferrum/client.rb

@@ -68,6 +68,7 @@ class Client
 
     def initialize(ws_url, options)
       @command_id = 0
+      @command_id_mutex = Mutex.new
       @ws_url = ws_url
       @options = options
       @pendings = Concurrent::Hash.new
@@ -154,8 +155,10 @@ def start
       end
     end
 
+    # Locked so two concurrent commands never share an id and read each
+    # other's responses from @pendings.
     def next_command_id
-      @command_id += 1
+      @command_id_mutex.synchronize { @command_id += 1 }
     end
 
     def raise_browser_error(error)

lib/ferrum/client/web_socket.rb

@@ -30,7 +30,10 @@ def initialize(url, max_receive_size, logger)
         end
 
         max_receive_size ||= ::WebSocket::Driver::MAX_LENGTH
-        @driver   = ::WebSocket::Driver.client(self, max_length: max_receive_size)
+        @driver = ::WebSocket::Driver.client(self, max_length: max_receive_size)
+        # websocket-driver holds no locks and is called from many threads: commands from
+        # callers, pong/close replies from the reader. One lock keeps frames from interleaving.
+        @driver_mutex = Mutex.new
         @messages = Queue.new
 
         @screenshot_commands = Concurrent::Hash.new if SKIP_LOGGING_SCREENSHOTS
@@ -41,7 +44,7 @@ def initialize(url, max_receive_size, logger)
 
         start
 
-        @driver.start
+        @driver_mutex.synchronize { @driver.start }
       end
 
       def on_open(_event)
@@ -76,7 +79,7 @@ def send_message(data)
         @screenshot_commands[data[:id]] = true if SKIP_LOGGING_SCREENSHOTS
 
         json = data.to_json
-        @driver.text(json)
+        @driver_mutex.synchronize { @driver.text(json) }
         @logger&.puts("\n\n▶ #{Utils::ElapsedTime.elapsed_time} #{json}")
       end
 
@@ -87,7 +90,7 @@ def write(data)
       end
 
       def close
-        @driver.close
+        @driver_mutex.synchronize { @driver.close }
       end
 
       private
@@ -98,7 +101,7 @@ def start
             data = @sock.readpartial(512)
             break unless data
 
-            @driver.parse(data)
+            @driver_mutex.synchronize { @driver.parse(data) }
           end
         rescue EOFError, Errno::ECONNRESET, Errno::EPIPE, IOError # rubocop:disable Lint/ShadowedException
           @messages.close

spec/unit/client/web_socket_spec.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+describe Ferrum::Client::WebSocket do
+  # `allocate` skips the constructor so no socket connects; the ivars the send
+  # path relies on are set by hand.
+  subject(:web_socket) { described_class.allocate }
+
+  # Stands in for the websocket driver and measures how many threads are inside
+  # `text` at once — anything above 1 corrupts the frame stream on a real driver.
+  let(:frame_probe_class) do
+    Class.new do
+      def initialize
+        @frames = Thread::Queue.new
+        @active = Concurrent::AtomicFixnum.new(0)
+        @peak = Concurrent::AtomicFixnum.new(0)
+      end
+
+      def text(json)
+        level = @active.increment
+        @peak.update { |peak| [peak, level].max }
+        sleep 0.005
+        @frames << json
+        @active.decrement
+      end
+
+      def frames = Array.new(@frames.size) { @frames.pop }
+
+      def peak_concurrent_writes = @peak.value
+    end
+  end
+
+  let(:frame_probe) { frame_probe_class.new }
+
+  before do
+    web_socket.instance_variable_set(:@driver, frame_probe)
+    web_socket.instance_variable_set(:@driver_mutex, Mutex.new)
+    web_socket.instance_variable_set(:@screenshot_commands, Concurrent::Hash.new)
+  end
+
+  describe "#send_message" do
+    it "hands the driver the message as JSON" do
+      web_socket.send_message(id: 1)
+      expect(frame_probe.frames).to eq(['{"id":1}'])
+    end
+
+    it "writes one frame at a time" do
+      Array.new(8) { |i| Thread.new { web_socket.send_message(id: i) } }.each(&:join)
+      expect(frame_probe.peak_concurrent_writes).to eq(1)
+    end
+  end
+end

spec/unit/client_spec.rb

@@ -0,0 +1,51 @@
+# frozen_string_literal: true
+
+describe Ferrum::Client do
+  # `allocate` skips the constructor so no browser boots; the ivars the id
+  # mint relies on are set by hand.
+  subject(:client) { described_class.allocate }
+
+  before do
+    client.instance_variable_set(:@command_id, 0)
+    client.instance_variable_set(:@command_id_mutex, Mutex.new)
+  end
+
+  describe "#build_message" do
+    it "carries the method and params" do
+      expect(client.build_message("Page.enable", foo: 1)).to include(method: "Page.enable", params: { foo: 1 })
+    end
+
+    it "mints sequential ids for sequential commands" do
+      ids = Array.new(3) { client.build_message("Page.enable", {})[:id] }
+      expect(ids).to eq([1, 2, 3])
+    end
+
+    context "with the id increment window held open" do
+      # Stretches the read-modify-write of `@command_id += 1` so the
+      # lost-update race is deterministic instead of scheduler-dependent.
+      let(:racy_counter_class) do
+        Class.new do
+          attr_reader :to_i
+
+          def initialize(value)
+            @to_i = value
+          end
+
+          def +(other)
+            read = @to_i
+            sleep 0.005
+            self.class.new(read + other)
+          end
+        end
+      end
+
+      before { client.instance_variable_set(:@command_id, racy_counter_class.new(0)) }
+
+      it "mints a unique id per command across concurrent threads" do
+        ids = Queue.new
+        Array.new(8) { Thread.new { ids << client.build_message("Page.enable", {})[:id].to_i } }.each(&:join)
+        expect(Array.new(8) { ids.pop }).to match_array((1..8).to_a)
+      end
+    end
+  end
+end