Aggregate and cleanup all CDP event threads on quit

When we quit driver, there should be no leftover event callback threads.
Otherwise, they would fail upon sending CDP command or waiting for its
response. Now the threads are aggregated to the thread group and killed
before closing the socket.

rb/lib/selenium/webdriver/devtools.rb

@@ -31,6 +31,7 @@ class DevTools
       autoload :Response, 'selenium/webdriver/devtools/response'
 
       def initialize(url:)
+        @callback_threads = ThreadGroup.new
         @messages = []
         @session_id = nil
         @url = url
@@ -41,6 +42,7 @@ def initialize(url:)
       end
 
       def close
+        @callback_threads.list.each(&:exit)
         socket.close
       end
 
@@ -93,27 +95,24 @@ def process_handshake
       end
 
       def attach_socket_listener
-        socket_listener = Thread.new do
+        Thread.new do
+          Thread.current.abort_on_exception = true
+          Thread.current.report_on_exception = false
+
           until socket.eof?
             incoming_frame << socket.readpartial(1024)
 
             while (frame = incoming_frame.next)
-              # Firefox will periodically fail on unparsable empty frame
-              break if frame.to_s.empty?
-
-              message = JSON.parse(frame.to_s)
-              @messages << message
-              WebDriver.logger.debug "DevTools <- #{message}"
+              message = process_frame(frame)
               next unless message['method']
 
+              params = message['params']
               callbacks[message['method']].each do |callback|
-                callback_thread = Thread.new(message['params'], &callback)
-                callback_thread.abort_on_exception = true
+                @callback_threads.add(callback_thread(params, &callback))
               end
             end
           end
         end
-        socket_listener.abort_on_exception = true
       end
 
       def start_session
@@ -127,6 +126,34 @@ def incoming_frame
         @incoming_frame ||= WebSocket::Frame::Incoming::Client.new(version: ws.version)
       end
 
+      def process_frame(frame)
+        message = frame.to_s
+
+        # Firefox will periodically fail on unparsable empty frame
+        return {} if message.empty?
+
+        message = JSON.parse(message)
+        @messages << message
+        WebDriver.logger.debug "DevTools <- #{message}"
+
+        message
+      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
+
+          yield params
+        end
+      end
+
       def wait
         @wait ||= Wait.new(timeout: RESPONSE_WAIT_TIMEOUT, interval: RESPONSE_WAIT_INTERVAL)
       end

rb/spec/integration/selenium/webdriver/devtools_spec.rb

@@ -22,9 +22,7 @@
 module Selenium
   module WebDriver
     describe DevTools, exclusive: {browser: %i[chrome edge firefox_nightly]} do
-      before(:all) { quit_driver }
-
-      after { quit_driver }
+      after { reset_driver! }
 
       it 'sends commands' do
         driver.devtools.page.navigate(url: url_for('xhtmlTest.html'))
@@ -42,6 +40,12 @@ module WebDriver
         expect(callback).to have_received(:call).at_least(:once)
       end
 
+      it 'propagates errors in events' do
+        driver.devtools.page.enable
+        driver.devtools.page.on(:load_event_fired) { raise "This is fine!" }
+        expect { driver.navigate.to url_for('xhtmlTest.html') }.to raise_error(RuntimeError, "This is fine!")
+      end
+
       context 'authentication', except: {browser: :firefox_nightly,
                                          reason: 'Fetch.enable is not yet supported'} do
         let(:username) { SpecSupport::RackServer::TestApp::BASIC_AUTH_CREDENTIALS.first }