Close CDP streams after reading

CHANGELOG.md

@@ -7,13 +7,15 @@
 - `page.accessibility` API and `Node#axnode` for reading the CDP accessibility tree
 
 ### Fixed
+- CDP streams opened for `Ferrum::Browser#pdf` and `Ferrum::Page::Tracing#record` are now closed with `IO.close` after being read.
 - `Ferrum::Node#type` / `Ferrum::Keyboard` now trigger the select-all editing shortcut (`Ctrl`/`Cmd`+`A`) by naming the CDP `commands` field, so selecting and replacing text in inputs and contenteditables works.
 - `Ferrum::Network::InterceptedRequest#match?` no longer coerces string blacklist/whitelist patterns containing regexp metacharacters
   (e.g. `?`, `.`) into regexps; string patterns are now compared literally instead. [#405], [#604]
 - `Ferrum::Network::AuthRequest#match?` and `Ferrum::Dialog#match?` received the same fix as [#604] above for consistency. String patterns
   now compare literally: exact-match for requests/auth requests, substring-match for dialog messages. [#405]
 
 ### Changed
+- An `IO.close` failure is now propagated after a PDF or tracing stream has otherwise been read successfully.
 - `Ferrum::PendingConnectionsError` and `Ferrum::TimeoutError` were swallowed even though happening when traffic iterator results in empty array. [#583]
 - `webrick` is no longer a runtime dependency. It is only required by `Ferrum::Proxy`, so add `gem "webrick"` to your Gemfile if you use it.
 

lib/ferrum/page/stream.rb

@@ -25,13 +25,25 @@ def stream_to_memory(encoding:, handle:)
       end
 
       def stream(output:, handle:)
+        completed = false
         loop do
           result = command("IO.read", handle: handle, size: STREAM_CHUNK)
           chunk = result.fetch("data")
           chunk = Base64.decode64(chunk) if result["base64Encoded"]
           output << chunk
           break if result["eof"]
         end
+        completed = true
+      ensure
+        close_stream(handle: handle, suppress_errors: !completed)
+      end
+
+      private
+
+      def close_stream(handle:, suppress_errors:)
+        command("IO.close", handle: handle)
+      rescue StandardError
+        raise unless suppress_errors
       end
     end
   end

spec/page/screenshot_spec.rb

@@ -429,6 +429,7 @@ def create_screenshot(path:, **options)
         allow(browser.page).to receive(:command).with("IO.read", hash_including(handle: "1")) {
           { "data" => "", "base64Encoded" => false, "eof" => true }
         }
+        allow(browser.page).to receive(:command).with("IO.close", handle: "1").and_return({})
 
         browser.pdf(path: file,
                     landscape: false,

spec/page/stream_spec.rb

@@ -0,0 +1,148 @@
+# frozen_string_literal: true
+
+describe Ferrum::Page::Stream do
+  subject(:streamer) { Object.new.extend(described_class) }
+
+  describe "#stream" do
+    let(:output) { String.new }
+    let(:handle) { "stream-handle" }
+
+    before do
+      allow(streamer).to receive(:command).with("IO.close", handle: handle).and_return({})
+    end
+
+    it "reads and appends chunks until EOF" do
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .twice
+        .and_return(
+          { "data" => "first", "base64Encoded" => false, "eof" => false },
+          { "data" => "second", "base64Encoded" => false, "eof" => true }
+        )
+
+      streamer.stream(output: output, handle: handle)
+
+      expect(output).to eq("firstsecond")
+    end
+
+    it "decodes Base64-encoded chunks" do
+      binary = "\x00\xFFpdf".b
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .once
+        .and_return(
+          { "data" => Base64.strict_encode64(binary), "base64Encoded" => true, "eof" => true }
+        )
+
+      streamer.stream(output: output, handle: handle)
+
+      expect(output).to eq(binary)
+    end
+
+    it "propagates errors raised while reading the stream" do
+      read_error = Ferrum::TimeoutError.new
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_raise(read_error)
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_return({})
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(Ferrum::TimeoutError) { |error| expect(error).to equal(read_error) }
+    end
+
+    it "closes the stream after reaching EOF" do
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_return({ "data" => "chunk", "base64Encoded" => false, "eof" => true })
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_return({})
+
+      streamer.stream(output: output, handle: handle)
+    end
+
+    it "closes the stream and propagates errors raised while writing output" do
+      write_error = IOError.new("write failed")
+      output = Object.new
+      allow(output).to receive(:<<).and_raise(write_error)
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_return({ "data" => "chunk", "base64Encoded" => false, "eof" => true })
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_return({})
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(IOError) { |error| expect(error).to equal(write_error) }
+    end
+
+    it "does not replace a read error when closing also fails" do
+      read_error = Ferrum::TimeoutError.new
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_raise(read_error)
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_raise(Ferrum::DeadBrowserError)
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(Ferrum::TimeoutError) { |error| expect(error).to equal(read_error) }
+    end
+
+    it "does not replace a write error when closing also fails" do
+      write_error = IOError.new("write failed")
+      output = Object.new
+      allow(output).to receive(:<<).and_raise(write_error)
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_return({ "data" => "chunk", "base64Encoded" => false, "eof" => true })
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_raise(Ferrum::DeadBrowserError)
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(IOError) { |error| expect(error).to equal(write_error) }
+    end
+
+    it "does not replace a non-StandardError when closing also fails" do
+      interrupt = Interrupt.new
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_raise(interrupt)
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_raise(Ferrum::DeadBrowserError)
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(Interrupt) { |error| expect(error).to equal(interrupt) }
+    end
+
+    it "propagates a close error after successfully reading the stream" do
+      close_error = Ferrum::DeadBrowserError.new
+      expect(streamer).to receive(:command)
+        .with("IO.read", handle: handle, size: described_class::STREAM_CHUNK)
+        .and_return({ "data" => "chunk", "base64Encoded" => false, "eof" => true })
+      expect(streamer).to receive(:command)
+        .with("IO.close", handle: handle)
+        .once
+        .and_raise(close_error)
+
+      expect do
+        streamer.stream(output: output, handle: handle)
+      end.to raise_error(Ferrum::DeadBrowserError) { |error| expect(error).to equal(close_error) }
+    end
+  end
+end