Close CDP streams after reading

CHANGELOG.md

@@ -7,6 +7,7 @@
   fixed port. Default (`wait: true`) keeps the previous synchronous behavior; `#restart` always waits.
 
 ### Changed
+- An `IO.close` failure is now propagated after a PDF or tracing stream has otherwise been read successfully.
 
 ### Fixed
 - `Ferrum::Contexts#connect_worker`/`#detach_unless_manually_attached` only rescued `Ferrum::BrowserError`, not `Ferrum::TimeoutError`.
@@ -27,6 +28,7 @@
   `--remote-debugging-address` and always binds/logs `127.0.0.1`, so Ferrum now substitutes the requested
   `:host` back into the address it uses to talk to Chrome (e.g. `Ferrum::Browser.new(host: "ferrum.localhost")`),
   instead of always hitting `127.0.0.1` [#552]
+- CDP streams opened for `Ferrum::Browser#pdf` and `Ferrum::Page::Tracing#record` are now closed with `IO.close` after being read.
 
 
 ### Removed

lib/ferrum/page/stream.rb

@@ -81,13 +81,25 @@ def stream_to_memory(encoding:, handle:)
       # @return [void]
       #
       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

@@ -430,6 +430,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