[rb] add synchronization and error handling for socket interactions

rb/lib/selenium/webdriver/bidi.rb

@@ -32,7 +32,7 @@ class BiDi
       autoload :InterceptedItem, 'selenium/webdriver/bidi/network/intercepted_item'
 
       def initialize(url:)
-        @ws = WebSocketConnection.new(url: url)
+        @ws = WebSocketConnection.new(url: url, protocol: :bidi)
       end
 
       def close

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

@@ -24,91 +24,128 @@ 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
       RESPONSE_WAIT_INTERVAL = 0.1
 
       MAX_LOG_MESSAGE_SIZE = 9999
 
-      def initialize(url:)
+      def initialize(url:, protocol: nil)
         @callback_threads = ThreadGroup.new
 
+        @callbacks_mtx = Mutex.new
+        @messages_mtx = Mutex.new
+        @closing_mtx = Mutex.new
+
+        @closing = false
         @session_id = nil
         @url = url
+        @protocol = protocol
 
         process_handshake
         @socket_thread = attach_socket_listener
       end
 
       def close
-        @callback_threads.list.each(&:exit)
-        @socket_thread.exit
-        socket.close
+        @closing_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 => e
+          WebDriver.logger.debug "Failed to join thread during close: #{e.class}: #{e.message}", id: :ws
+        end
       end
 
       def callbacks
         @callbacks ||= Hash.new { |callbacks, event| callbacks[event] = [] }
       end
 
       def add_callback(event, &block)
-        callbacks[event] << block
-        block.object_id
+        @callbacks_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 }
+        @callbacks_mtx.synchronize do
+          return if @closing
+
+          callbacks_for_event = callbacks[event]
+          return if callbacks_for_event.reject! { |cb| cb.object_id == id }
 
-        ids = callbacks[event]&.map(&:object_id)
-        raise Error::WebDriverError, "Callback with ID #{id} does not exist for event #{event}: #{ids}"
+          ids = callbacks_for_event.map(&:object_id)
+          raise Error::WebDriverError, "Callback with ID #{id} does not exist for event #{event}: #{ids}"
+        end
       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 e, "WebSocket is closed (#{e.class}: #{e.message})"
+        end
+
+        wait.until { @messages_mtx.synchronize { messages.delete(id) } }
       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|
-                @callback_threads.add(callback_thread(params, &callback))
+              @messages_mtx.synchronize { callbacks[message['method']].dup }.each do |callback|
+                @callback_threads.add(callback_thread(message['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 +159,33 @@ 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)
+        @messages_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
+          return if @closing
+
+          if devtools?
+            # Async thread exceptions are not deterministic and should not be relied on; we should stop
+            WebDriver.logger.deprecate('propogating errors from DevTools callbacks')
+            Thread.main.raise(e)
+          end
+
+          bt = Array(e.backtrace).first(5).join("\n")
+          WebDriver.logger.error "Callback error: #{e.class}: #{e.message}\n#{bt}", id: :ws
         end
       end
 
@@ -167,6 +210,14 @@ def ws
         @ws ||= WebSocket::Handshake::Client.new(url: @url)
       end
 
+      def devtools?
+        @protocol == :devtools
+      end
+
+      def bidi?
+        @protocol == :bidi
+      end
+
       def next_id
         @id ||= 0
         @id += 1

rb/lib/selenium/webdriver/devtools.rb

@@ -29,7 +29,7 @@ class DevTools
       autoload :Response, 'selenium/webdriver/devtools/response'
 
       def initialize(url:, target_type:)
-        @ws = WebSocketConnection.new(url: url)
+        @ws = WebSocketConnection.new(url: url, protocol: :devtools)
         @session_id = nil
         start_session(target_type: target_type)
       end

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

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

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

@@ -206,12 +206,20 @@ 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
+          begin
+            execute :delete_session
+          rescue *QUIT_ERRORS => e
+            WebDriver.logger.debug "delete_session failed during quit: #{e.class}: #{e.message}", id: :quit
+          ensure
+            begin
+              http.close
+            rescue *QUIT_ERRORS => e
+              WebDriver.logger.debug "http.close failed during quit: #{e.class}: #{e.message}", id: :quit
+            end
+          end
           nil
         end
 

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

@@ -5,6 +5,8 @@ module Selenium
 
       @callback_threads: untyped
 
+      @protocol: Symbol?
+
       @session_id: untyped
 
       @url: untyped
@@ -33,7 +35,7 @@ module Selenium
 
       MAX_LOG_MESSAGE_SIZE: Integer
 
-      def initialize: (url: untyped) -> void
+      def initialize: (url: String, ?protocol?: Symbol) -> void
 
       def add_callback: (untyped event) { () -> void } -> untyped
 
@@ -47,6 +49,10 @@ module Selenium
 
       private
 
+      def bidi?: -> bool
+
+      def devtools?: -> bool
+
       def messages: () -> untyped
 
       def process_handshake: () -> untyped

rb/spec/integration/selenium/webdriver/remote/driver_spec.rb

@@ -83,8 +83,7 @@ module Remote
           end
         end
 
-        it 'errors when not set', {except: {browser: :firefox, reason: 'grid always sets true and firefox returns it'},
-                                   exclude: {browser: :safari, reason: 'grid hangs'}} do
+        it 'errors when not set', exclude: {browser: :safari, reason: 'grid hangs'} do
           reset_driver!(enable_downloads: false) do |driver|
             expect {
               driver.downloadable_files