Fix ESRIGeoJSON: handle Empty reply from server and paginate past maxRecordCount

lib/spatial_features.rb

@@ -10,6 +10,7 @@
 require 'spatial_features/uncached_result'
 require 'spatial_features/venn_polygons'
 require 'spatial_features/controller_helpers/spatial_extensions'
+require 'spatial_features/gdal'
 require 'spatial_features/download'
 require 'spatial_features/unzip'
 require 'spatial_features/utils'

lib/spatial_features/download.rb

@@ -1,42 +1,93 @@
+require 'net/http'
 require 'open-uri'
+require 'openssl'
 
 module SpatialFeatures
   module Download
-    # file can be a url, path, or file, any of which can return be a zipped archive
-    def self.open(file)
-      file = URI.open(file)
-      file = normalize_file(file) if file.is_a?(StringIO)
-      return file
-    end
+    REMOTE_URL = %r{\Ahttps?://}i.freeze
+
+    # Seconds to wait for a remote source, applied both to establishing the connection and to
+    # each read. Without it a hung server holds an import worker open indefinitely.
+    mattr_accessor :timeout
+    self.timeout = 60
 
-    # file can be a url, path, or file, any of which can return be a zipped archive
-    def self.open_each(path_or_url, unzip: nil, **unzip_options)
-      file = Download.open(path_or_url)
-      files = if unzip && Unzip.is_zip?(file)
-        find_in_zip(file, find: unzip, **unzip_options)
-      else
-        [file]
+    # Errors meaning a remote source could not be read at all, as opposed to being read and
+    # found unusable. `OpenURI::HTTPError` covers the 4xx and 5xx replies.
+    UNREACHABLE_ERRORS = [::OpenURI::HTTPError, ::SocketError, ::SystemCallError,
+                          ::Net::OpenTimeout, ::Net::ReadTimeout, ::OpenSSL::SSL::SSLError].freeze
+
+    class << self
+      # Returns an open File for `file`, which may be a URL, a path, or a File. The content may
+      # be a zipped archive; `::open_each` unwraps one.
+      #
+      # @raise [SpatialFeatures::ImportError] when a remote source cannot be reached.
+      def open(file)
+        file = fetch(file)
+        file = normalize_file(file) if file.is_a?(StringIO)
+        return file
       end
 
-      return files.map { |f| File.open(f) }
-    end
+      # Returns the body of `path_or_url` as a String, without writing it to disk.
+      #
+      # @raise [SpatialFeatures::ImportError] when a remote source cannot be reached.
+      def read(path_or_url)
+        fetch(path_or_url).read
+      end
+
+      # Returns an open File for each source in `path_or_url`, unwrapping an archive when
+      # `unzip` is given a pattern its entries can match.
+      def open_each(path_or_url, unzip: nil, **unzip_options)
+        file = Download.open(path_or_url)
+        files = if unzip && Unzip.is_zip?(file)
+          find_in_zip(file, find: unzip, **unzip_options)
+        else
+          [file]
+        end
 
-    def self.normalize_file(file)
-      Tempfile.new.tap do |temp|
-        temp.binmode
-        temp.write(file.read)
-        temp.rewind
+        return files.map { |f| File.open(f) }
       end
-    end
 
-    def self.entries(file)
-      file = Kernel.open(file)
-      file = normalize_file(file) if file.is_a?(StringIO)
-      Unzip.entries(file)
-    end
+      def normalize_file(file)
+        Tempfile.new.tap do |temp|
+          temp.binmode
+          temp.write(file.read)
+          temp.rewind
+        end
+      end
+
+      # Returns the entries of the archive at `file` without extracting them.
+      def entries(file)
+        file = fetch(file)
+        file = normalize_file(file) if file.is_a?(StringIO)
+        Unzip.entries(file)
+      end
+
+      def find_in_zip(file, find:, **unzip_options)
+        Unzip.paths(file, find: find, **unzip_options)
+      end
+
+      private
 
-    def self.find_in_zip(file, find:, **unzip_options)
-      Unzip.paths(file, find: find, **unzip_options)
+      # Returns an IO for `file`: fetched over the network when it is a remote URL, opened from
+      # disk when it is any other String, and left to open itself otherwise.
+      #
+      # @note A local path goes to `File.open`, never `URI.open`. `URI.open` hands anything
+      #   that is not a URL to `Kernel#open`, which runs the name as a command when it begins
+      #   with a pipe.
+      # @note Timeouts and the unreachable rescue apply only to a remote URL. `URI.open`
+      #   rejects the timeouts when handed an already open file, and `Errno::ENOENT` for a
+      #   local path is a `SystemCallError` that callers turn into a message for the person who
+      #   uploaded the file, without naming the path the server looked in.
+      def fetch(file)
+        return URI.open(file) unless file.is_a?(String)
+        return File.open(file) unless file.match?(REMOTE_URL)
+
+        begin
+          URI.open(file, :open_timeout => timeout, :read_timeout => timeout)
+        rescue *UNREACHABLE_ERRORS => e
+          raise SpatialFeatures::ImportError, "This source could not be reached. #{e.message}"
+        end
+      end
     end
   end
 end

lib/spatial_features/gdal.rb

@@ -0,0 +1,42 @@
+require 'open3'
+
+module SpatialFeatures
+  # Runs the GDAL command line tools.
+  #
+  # Every argument reaches the tool as a single argv entry, so a path or a projection taken
+  # from an uploaded archive arrives as one string rather than as shell syntax.
+  module GDAL
+    class << self
+      # Returns what `tool` wrote to standard output.
+      #
+      # @param tool [String] the executable name, such as `ogr2ogr`.
+      # @param args [Array<String>] one argv entry each.
+      # @return [String] the output, empty when the tool wrote nothing.
+      # @raise [Errno::ENOENT] when the tool is not installed.
+      def capture(tool, *args)
+        Open3.capture2(*argv(tool, args)).first
+      end
+
+      # Runs `tool` for its exit status rather than its output.
+      #
+      # @param tool [String] the executable name, such as `ogr2ogr`.
+      # @param args [Array<String>] one argv entry each.
+      # @return [Boolean] true when the tool exited successfully.
+      # @raise [Errno::ENOENT] when the tool is not installed.
+      def run(tool, *args)
+        system(*argv(tool, args))
+      end
+
+      private
+
+      # Returns the argv to spawn `tool` with.
+      #
+      # @note The command is a two element array so that Ruby spawns the executable directly.
+      #   Both `system` and `Open3` fall back to a shell when handed a lone string, which
+      #   would put every argument here back in reach of shell parsing.
+      def argv(tool, args)
+        [[tool.to_s, tool.to_s], *args.map(&:to_s)]
+      end
+    end
+  end
+end

lib/spatial_features/importers/esri_geo_json.rb

@@ -1,10 +1,18 @@
 require 'digest/md5'
-require 'open3'
+require 'json'
+require 'tempfile'
 require 'spatial_features/importers/geo_json'
 
 module SpatialFeatures
   module Importers
     class ESRIGeoJSON < GeoJSON
+      # How many pages to walk before concluding the endpoint is ignoring `resultOffset` and
+      # serving the same features over and over. A service capped at the usual 1000 or 2000
+      # features per page would have to hold half a million features to reach this
+      # legitimately. Raise it for a layer that genuinely holds more.
+      class_attribute :max_pages
+      self.max_pages = 500
+
       def parsed_geojson
         @parsed_geojson ||= JSON.parse(geojson)
       end
@@ -15,10 +23,115 @@ def geojson
 
       private
 
-      def esri_json_to_geojson(url)
-        args = ['ogr2ogr', '-t_srs', 'EPSG:4326', '-f', 'GeoJSON', '/dev/stdout', url]
-        args << 'OGRGeoJSON' unless URI.parse(url).relative? # A relative URL is a local file path
-        Open3.capture2(*args).first
+      # Returns the layer as GeoJSON. A relative path is read from disk; anything else is
+      # downloaded first, so OGR is always given a local file.
+      #
+      # @param path_or_url [String] a local file path, or the URL of an ArcGIS query endpoint.
+      # @return [String] a GeoJSON FeatureCollection in EPSG:4326.
+      # @note Downloading rather than passing the URL to OGR is what makes an endpoint offering
+      #   only HTTPS 1.1 readable. GDAL's curl client gets an empty reply from those servers,
+      #   which surfaces as `ERROR 1: Empty reply from server`.
+      def esri_json_to_geojson(path_or_url)
+        return ogr2ogr_to_geojson(path_or_url) if URI.parse(path_or_url).relative?
+
+        with_downloaded_file(path_or_url) do |path|
+          ogr2ogr_to_geojson(path)
+        end
+      end
+
+      # Returns the GeoJSON OGR reads out of the file at `path`, reprojected to EPSG:4326.
+      # OGR selects its driver by inspecting the content, so the file may hold either GeoJSON
+      # or ESRI JSON.
+      #
+      # @note No layer name is passed. OGR names a local file's layer after its basename, and
+      #   naming a layer that does not exist fails the read.
+      def ogr2ogr_to_geojson(path)
+        GDAL.capture('ogr2ogr', '-t_srs', 'EPSG:4326', '-f', 'GeoJSON', '/dev/stdout', path)
+      end
+
+      # Downloads the query into a tempfile and yields its path, removing it afterwards.
+      def with_downloaded_file(url)
+        Tempfile.create(['esri_geojson', '.json']) do |tempfile|
+          tempfile.binmode
+          download_paginated(url, tempfile)
+          tempfile.close
+          return yield(tempfile.path)
+        end
+      end
+
+      # Walks the query's pages with `resultOffset` and writes them to `io` as one collection.
+      # ArcGIS endpoints cap each response at the service's `maxRecordCount`, commonly 1000 or
+      # 2000 features, and set `exceededTransferLimit` while more results are waiting.
+      #
+      # @note Raises `SpatialFeatures::ImportError` once `max_pages` requests have been made
+      #   and the endpoint still reports more, since a server that ignores `resultOffset`
+      #   otherwise repeats its first page until the process runs out of memory.
+      def download_paginated(url, io)
+        combined = nil
+        offset = 0
+        pages = 0
+
+        loop do
+          page = fetch_page(paginated_url(url, offset))
+          page_features = page['features'] || []
+
+          if combined.nil?
+            combined = page
+          else
+            combined['features'].concat(page_features)
+          end
+
+          break if page_features.empty? || !exceeded_transfer_limit?(page)
+
+          pages += 1
+          if pages >= max_pages
+            raise SpatialFeatures::ImportError,
+                  "This layer was still reporting more features after #{max_pages} requests. " \
+                  "The server may be ignoring the `resultOffset` parameter."
+          end
+
+          offset += page_features.length
+        end
+
+        if combined
+          combined.delete('exceededTransferLimit')
+          combined['properties']&.delete('exceededTransferLimit')
+          io.write(JSON.dump(combined))
+        end
+      end
+
+      # Returns one page of the query.
+      #
+      # @return [Hash] the parsed response body.
+      # @raise [SpatialFeatures::ImportError] when the endpoint cannot be reached, or answers
+      #   with a body that is not JSON. An endpoint that rejects a query replies with HTML or
+      #   an error document, which the parse failure alone does not convey.
+      def fetch_page(url)
+        body = Download.read(url)
+        JSON.parse(body)
+      rescue JSON::ParserError
+        raise SpatialFeatures::ImportError,
+              "This layer did not return map data. The server replied with #{body.to_s[0, 100].inspect}."
+      end
+
+      # Returns true while the service reports that more features are waiting. Services set the
+      # flag at the top level or under `properties` depending on the response format.
+      def exceeded_transfer_limit?(page)
+        page['exceededTransferLimit'] || page.dig('properties', 'exceededTransferLimit')
+      end
+
+      # Returns `url` with `resultOffset` set to `offset`, replacing any the caller supplied.
+      # Returns it unchanged for the first page, so a service that does not paginate is asked
+      # exactly what the caller asked for.
+      def paginated_url(url, offset)
+        return url if offset.zero?
+
+        uri = URI.parse(url)
+        params = URI.decode_www_form(uri.query || '')
+        params.reject! { |key, _| key == 'resultOffset' }
+        params << ['resultOffset', offset.to_s]
+        uri.query = URI.encode_www_form(params)
+        uri.to_s
       end
     end
   end

lib/spatial_features/importers/shapefile.rb

@@ -1,6 +1,5 @@
 require 'ostruct'
 require 'digest/md5'
-require 'open3'
 
 module SpatialFeatures
   module Importers
@@ -101,7 +100,7 @@ def validate_shapefile!(file_path)
       def project_to_4326(file_path)
         output_path = Tempfile.create([::File.basename(file_path, '.shp') + '_epsg_4326_', '.shp']) { |file| file.path }
         return unless (proj4 = proj4_from_file(file_path))
-        return unless system('ogr2ogr', '-s_srs', proj4, '-t_srs', 'EPSG:4326', output_path, file_path)
+        return unless GDAL.run('ogr2ogr', '-s_srs', proj4, '-t_srs', 'EPSG:4326', output_path, file_path)
         return ::File.open(output_path)
       end
 
@@ -112,7 +111,7 @@ def proj4_from_file(file_path)
         # Sanitize: "'+proj=utm +zone=11 +datum=NAD83 +units=m +no_defs '\n" and lately
         #           "+proj=utm +zone=11 +datum=NAD83 +units=m +no_defs \n" to
         #           "+proj=utm +zone=11 +datum=NAD83 +units=m +no_defs"
-        Open3.capture2('gdalsrsinfo', file_path, '-o', 'proj4').first.strip.remove(/^'|'$/).presence
+        GDAL.capture('gdalsrsinfo', file_path, '-o', 'proj4').strip.remove(/^'|'$/).presence
       rescue Errno::ENOENT
         nil
       end