[rb] add synchronization and error handling for socket interactions

Author: titusfortner

rb/lib/selenium/webdriver/common/websocket_connection.rb

@@ -24,7 +24,10 @@ module WebDriver
     class WebSocketConnection
       CONNECTION_ERRORS = [
         Errno::ECONNRESET, # connection is aborted (browser process was killed)
-        Errno::EPIPE # broken pipe (browser process was killed)
+        Errno::EPIPE, # broken pipe (browser process was killed)
+        Errno::EBADF, # file descriptor already closed (double-close or GC)
+        IOError, # Ruby socket read/write after close
+        EOFError # socket reached EOF after remote closed cleanly
       ].freeze
 
       RESPONSE_WAIT_TIMEOUT = 30
@@ -34,7 +37,8 @@ class WebSocketConnection
 
       def initialize(url:)
         @callback_threads = ThreadGroup.new
-
+        @mtx = Mutex.new
+        @closing = false
         @session_id = nil
         @url = url
 
@@ -43,72 +47,99 @@ def initialize(url:)
       end
 
       def close
-        @callback_threads.list.each(&:exit)
-        @socket_thread.exit
-        socket.close
+        @mtx.synchronize do
+          return if @closing
+
+          @closing = true
+        end
+
+        begin
+          socket.close
+        rescue *CONNECTION_ERRORS => e
+          WebDriver.logger.debug "WebSocket listener closed: #{e.class}: #{e.message}", id: :ws
+          # already closed
+        end
+
+        # Let threads unwind instead of calling exit
+        @socket_thread&.join(0.5)
+        @callback_threads.list.each do |thread|
+          thread.join(0.5)
+        rescue StandardError
+          nil
+        end
       end
 
       def callbacks
         @callbacks ||= Hash.new { |callbacks, event| callbacks[event] = [] }
       end
 
       def add_callback(event, &block)
-        callbacks[event] << block
-        block.object_id
+        @mtx.synchronize do
+          callbacks[event] << block
+          block.object_id
+        end
       end
 
       def remove_callback(event, id)
-        return if callbacks[event].reject! { |callback| callback.object_id == id }
+        removed = @mtx.synchronize { callbacks[event].reject! { |cb| cb.object_id == id } }
+        return if removed || @closing
 
-        ids = callbacks[event]&.map(&:object_id)
+        ids = @mtx.synchronize { callbacks[event]&.map(&:object_id) }
         raise Error::WebDriverError, "Callback with ID #{id} does not exist for event #{event}: #{ids}"
       end
 
       def send_cmd(**payload)
         id = next_id
         data = payload.merge(id: id)
-        WebDriver.logger.debug "WebSocket -> #{data}"[...MAX_LOG_MESSAGE_SIZE], id: :bidi
+        WebDriver.logger.debug "WebSocket -> #{data}"[...MAX_LOG_MESSAGE_SIZE], id: :ws
         data = JSON.generate(data)
         out_frame = WebSocket::Frame::Outgoing::Client.new(version: ws.version, data: data, type: 'text')
-        socket.write(out_frame.to_s)
 
-        wait.until { messages.delete(id) }
+        begin
+          socket.write(out_frame.to_s)
+        rescue *CONNECTION_ERRORS => e
+          raise Error::WebDriverError, "WebSocket is closed (#{e.class}: #{e.message})"
+        end
+
+        wait.until do
+          @mtx.synchronize { messages.delete(id) }
+        end
       end
 
       private
 
-      # We should be thread-safe to use the hash without synchronization
-      # because its keys are WebSocket message identifiers and they should be
-      # unique within a devtools session.
       def messages
         @messages ||= {}
       end
 
       def process_handshake
         socket.print(ws.to_s)
-        ws << socket.readpartial(1024)
+        ws << socket.readpartial(1024) until ws.finished?
       end
 
       def attach_socket_listener
         Thread.new do
-          Thread.current.abort_on_exception = true
           Thread.current.report_on_exception = false
 
-          until socket.eof?
+          loop do
+            break if @closing
+
             incoming_frame << socket.readpartial(1024)
 
             while (frame = incoming_frame.next)
+              break if @closing
+
               message = process_frame(frame)
               next unless message['method']
 
               params = message['params']
-              callbacks[message['method']].each do |callback|
+              @mtx.synchronize { callbacks[message['method']].dup }.each do |callback|
                 @callback_threads.add(callback_thread(params, &callback))
               end
             end
           end
-        rescue *CONNECTION_ERRORS
-          Thread.stop
+        rescue *CONNECTION_ERRORS, WebSocket::Error => e
+          WebDriver.logger.debug "WebSocket listener closed: #{e.class}: #{e.message}", id: :ws
         end
       end
 
@@ -122,27 +153,26 @@ def process_frame(frame)
         # Firefox will periodically fail on unparsable empty frame
         return {} if message.empty?
 
-        message = JSON.parse(message)
-        messages[message['id']] = message
-        WebDriver.logger.debug "WebSocket <- #{message}"[...MAX_LOG_MESSAGE_SIZE], id: :bidi
+        msg = JSON.parse(message)
+        @mtx.synchronize { messages[msg['id']] = msg if msg.key?('id') }
 
-        message
+        WebDriver.logger.debug "WebSocket <- #{msg}"[...MAX_LOG_MESSAGE_SIZE], id: :ws
+        msg
       end
 
       def callback_thread(params)
         Thread.new do
-          Thread.current.abort_on_exception = true
-
-          # We might end up blocked forever when we have an error in event.
-          # For example, if network interception event raises error,
-          # the browser will keep waiting for the request to be proceeded
-          # before returning back to the original thread. In this case,
-          # we should at least print the error.
-          Thread.current.report_on_exception = true
+          Thread.current.abort_on_exception = false
+          Thread.current.report_on_exception = false
+          return if @closing
 
           yield params
-        rescue Error::WebDriverError, *CONNECTION_ERRORS
-          Thread.stop
+        rescue Error::WebDriverError, *CONNECTION_ERRORS => e
+          WebDriver.logger.debug "Callback aborted: #{e.class}: #{e.message}", id: :ws
+        rescue StandardError => e
+          # Unexpected handler failure; log with a short backtrace.
+          bt = Array(e.backtrace).first(5).join("\n")
+          WebDriver.logger.error "Callback error: #{e.class}: #{e.message}\n#{bt}", id: :ws
         end
       end
 

rb/lib/selenium/webdriver/remote/bidi_bridge.rb

@@ -46,9 +46,10 @@ def refresh
         end
 
         def quit
-          super
-        ensure
           bidi.close
+          super
+        rescue *QUIT_ERRORS
+          nil
         end
 
         def close

rb/lib/selenium/webdriver/remote/bridge.rb

@@ -206,13 +206,16 @@ def switch_to_default_content
           switch_to_frame nil
         end
 
-        QUIT_ERRORS = [IOError].freeze
+        QUIT_ERRORS = [IOError, EOFError, WebSocket::Error].freeze
 
         def quit
           execute :delete_session
-          http.close
-        rescue *QUIT_ERRORS
-          nil
+        ensure
+          begin
+            http.close
+          rescue *QUIT_ERRORS
+            nil
+          end
         end
 
         def close