Send :read_timeout when streaming events (#584)

- Corrects a bug in `Docker::Connection.compile_request_params` that incorrectly stripped out `read_timeout` when it was nil.
- Sets `Docker::Event.stream` to skip timeouts by default.
- Sets `Docker::Event.stream` to avoid automatically retrying timeout-related events by default.
- Adds timeout-related event streaming tests. A particularly slow (60s+) test can be enabled by setting `RUN_SLOW_TESTS=1` in the testing environment.

Author: Daniel Schmidt

lib/docker/connection.rb

@@ -137,24 +137,24 @@ def version
   end
 
 private
-  # Given an HTTP method, path, optional query, extra options, and block,
-  # compiles a request.
+
   def compile_request_params(http_method, path, query = nil, opts = nil, &block)
-    query ||= {}
     opts ||= {}
-    headers = opts.delete(:headers) || {}
-    content_type = opts[:body].nil? ?  'text/plain' : 'application/json'
-    user_agent = "Swipely/Docker-API #{Docker::VERSION}"
+    query ||= opts.delete(:query) || {}
+
+    default_headers = {
+      'Content-Type' => opts[:body].nil? ? 'text/plain' : 'application/json',
+      'User-Agent' => "Swipely/Docker-API #{Docker::VERSION}",
+    }
+    headers = default_headers.merge(opts.delete(:headers) || {})
+
     {
-      :method        => http_method,
-      :path          => path,
-      :query         => query,
-      :headers       => { 'Content-Type' => content_type,
-                          'User-Agent'   => user_agent,
-                        }.merge(headers),
-      :expects       => (200..204).to_a << 301 << 304,
-      :idempotent    => http_method == :get,
-      :request_block => block,
-    }.merge(opts).reject { |_, v| v.nil? }
+      method: http_method,
+      path: path,
+      headers:,
+      query:,
+      expects: (200..204).to_a << 301 << 304,
+      idempotent: http_method == :get,
+    }.merge(opts).tap { |params| params[:request_block] = block if block }
   end
 end

lib/docker/event.rb

@@ -30,11 +30,21 @@ class << self
     include Docker::Error
 
     def stream(opts = {}, conn = Docker.connection, &block)
-      conn.get('/events', opts, :response_block => lambda { |b, r, t|
-        b.each_line do |line|
-          block.call(new_event(line, r, t))
+      # Disable timeouts by default
+      opts[:read_timeout] = nil unless opts.key? :read_timeout
+
+      # By default, avoid retrying timeout errors. Set opts[:retry_errors] to override this.
+      opts[:retry_errors] ||= Excon::DEFAULT_RETRY_ERRORS.reject do |cls|
+        cls == Excon::Error::Timeout
+      end
+
+      opts[:response_block] = lambda do |chunk, remaining, total|
+        chunk.each_line do |event_json|
+          block.call(new_event(event_json, remaining, total))
         end
-      })
+      end
+
+      conn.get('/events', opts.delete(:query), opts)
     end
 
     def since(since, opts = {}, conn = Docker.connection, &block)

spec/docker/event_spec.rb

@@ -1,6 +1,7 @@
 # frozen_string_literal: true
 
 require 'spec_helper'
+require 'securerandom'
 
 SingleCov.covered! uncovered: 5
 
@@ -87,6 +88,43 @@
 
       container.remove
     end
+
+    context 'with timeouts' do
+      # @see https://github.com/upserve/docker-api/issues/584
+
+      it 'does not (seem to) time out by default' do
+        # @note Excon passes timeout-related arguments directly to IO.select, which in turn makes a system call,
+        #   so it's not possible to mock this (e.g. w/Timecop).
+        skip_slow_test
+        expect { Timeout.timeout(65) { stream_events_with_timeout } }
+          .to raise_error Timeout::Error
+      end
+
+      it 'times out immediately if read_timeout < Timeout.timeout' do
+        expect { Timeout.timeout(1) { stream_events_with_timeout(0) } }
+          .to raise_error Docker::Error::TimeoutError
+      end
+
+      it 'times out after timeout(1) if read_timeout=2' do
+        expect { Timeout.timeout(1) { stream_events_with_timeout(2) } }
+          .to raise_error Timeout::Error
+      end
+
+      private
+
+      def stream_events_with_timeout(read_timeout = [])
+        opts = {
+          # Filter to avoid unexpected Docker events interfering with timeout behavior
+          query: { filters: { container: [SecureRandom.uuid] }.to_json },
+          # Use [] to differentiate between explicit nil and not providing an arg (falling back to the default)
+          read_timeout:,
+        }.reject { |_, v| v.empty? rescue false }
+
+        Docker::Event.stream(opts) do |event|
+          raise "Unexpected event interfered with timeout test: #{event}"
+        end
+      end
+    end
   end
 
   describe ".since" do

spec/spec_helper.rb

@@ -24,6 +24,10 @@ def project_dir
 end
 
 module SpecHelpers
+  def skip_slow_test
+    skip "Disabled because ENV['RUN_SLOW_TESTS'] not set" unless ENV['RUN_SLOW_TESTS']
+  end
+
   def skip_without_auth
     skip "Disabled because of missing auth" if ENV['DOCKER_API_USER'] == 'debbie_docker'
   end