Stream tarball downloads and file uploads to avoid memory buffering

lib/hexdocs/application.ex

@@ -12,6 +12,7 @@ defmodule Hexdocs.Application do
     Logger.info("Running Cowboy with #{inspect(cowboy_options)}")
 
     children = [
+      Hexdocs.TmpDir,
       {Task.Supervisor, name: Hexdocs.Tasks},
       {Hexdocs.Debouncer, name: Hexdocs.Debouncer},
       goth_spec(),

lib/hexdocs/bucket.ex

@@ -23,8 +23,6 @@ defmodule Hexdocs.Bucket do
       meta: [{"surrogate-key", key}]
     ]
 
-    Logger.info("Uploading docs_public_bucket #{path}")
-
     case Hexdocs.Store.put(:docs_public_bucket, path, content, opts) do
       {:ok, 200, _headers, _body} ->
         :ok
@@ -39,10 +37,10 @@ defmodule Hexdocs.Bucket do
     purge([key])
   end
 
-  def upload(repository, package, version, all_versions, files) do
+  def upload(repository, package, version, all_versions, dir, files) do
     latest_version? = Hexdocs.Utils.latest_version?(package, version, all_versions)
     upload_type = upload_type(latest_version?)
-    upload_files = list_upload_files(repository, package, version, files, upload_type)
+    upload_files = list_upload_files(repository, package, version, dir, files, upload_type)
     paths = MapSet.new(upload_files, &elem(&1, 0))
 
     upload_new_files(upload_files)
@@ -53,7 +51,7 @@ defmodule Hexdocs.Bucket do
       {:docs_config, repository, package},
       @gcs_put_debounce,
       fn ->
-        docs_config = build_docs_config(repository, package, version, all_versions, files)
+        docs_config = build_docs_config(repository, package, version, all_versions, dir, files)
         upload_new_files([docs_config])
       end
     )
@@ -63,17 +61,24 @@ defmodule Hexdocs.Bucket do
   end
 
   # For Elixir and Hex we use the docs_config.js included in the tarball
-  defp build_docs_config(repository, package, _version, _all_versions, files)
+  defp build_docs_config(repository, package, _version, _all_versions, dir, files)
        when package in @special_package_names do
     path = "docs_config.js"
     unversioned_path = repository_path(repository, Path.join([package, path]))
     cdn_key = docs_config_cdn_key(repository, package)
-    {"docs_config.js", data} = List.keyfind(files, "docs_config.js", 0)
+
+    data =
+      if "docs_config.js" in files do
+        File.read!(Path.join(dir, "docs_config.js"))
+      else
+        ""
+      end
+
     {unversioned_path, cdn_key, data, public?(repository)}
   end
 
   # TODO: don't include retired versions?
-  defp build_docs_config(repository, package, version, all_versions, _files) do
+  defp build_docs_config(repository, package, version, all_versions, _dir, _files) do
     versions =
       if version in all_versions do
         all_versions
@@ -135,22 +140,32 @@ defmodule Hexdocs.Bucket do
     cond do
       deleting_latest_version? && new_latest_version ->
         key = build_key(repository, package, new_latest_version)
-        body = Hexdocs.Store.get(:repo_bucket, key)
-
-        case Hexdocs.Tar.unpack(body, repository: repository, package: package, version: version) do
-          {:ok, files} ->
-            upload_files =
-              list_upload_files(repository, package, new_latest_version, files, :both)
-
-            paths = MapSet.new(upload_files, &elem(&1, 0))
-            update_versions = [version, new_latest_version]
-
-            upload_new_files(upload_files)
-            delete_old_docs(repository, package, update_versions, paths, :both)
-            purge_hexdocs_cache(repository, package, update_versions, :both)
-
-          {:error, reason} ->
-            Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+        tarball_path = Hexdocs.TmpDir.tmp_file("docs-tarball")
+
+        case Hexdocs.Store.get_to_file(:repo_bucket, key, tarball_path) do
+          :ok ->
+            case Hexdocs.Tar.unpack_to_dir({:file, tarball_path},
+                   repository: repository,
+                   package: package,
+                   version: version
+                 ) do
+              {:ok, dir, files} ->
+                upload_files =
+                  list_upload_files(repository, package, new_latest_version, dir, files, :both)
+
+                paths = MapSet.new(upload_files, &elem(&1, 0))
+                update_versions = [version, new_latest_version]
+
+                upload_new_files(upload_files)
+                delete_old_docs(repository, package, update_versions, paths, :both)
+                purge_hexdocs_cache(repository, package, update_versions, :both)
+
+              {:error, reason} ->
+                Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+            end
+
+          nil ->
+            Logger.error("Failed to get tarball #{repository}/#{package} #{new_latest_version}")
         end
 
       deleting_latest_version? ->
@@ -171,21 +186,23 @@ defmodule Hexdocs.Bucket do
     Path.join(["repos", repository, "docs", "#{package}-#{version}.tar.gz"])
   end
 
-  defp list_upload_files(repository, package, version, files, upload_type) do
+  defp list_upload_files(repository, package, version, dir, files, upload_type) do
     Enum.flat_map(files, fn
-      {"docs_config.js", _data} ->
+      "docs_config.js" ->
         []
 
-      {path, data} ->
+      path ->
+        source = Path.join(dir, path)
+
         versioned_path =
           repository_path(repository, Path.join([package, to_string(version), path]))
 
         cdn_key = docspage_versioned_cdn_key(repository, package, version)
-        versioned = {versioned_path, cdn_key, data, public?(repository)}
+        versioned = {versioned_path, cdn_key, {:file, source}, public?(repository)}
 
         unversioned_path = repository_path(repository, Path.join([package, path]))
         cdn_key = docspage_unversioned_cdn_key(repository, package)
-        unversioned = {unversioned_path, cdn_key, data, public?(repository)}
+        unversioned = {unversioned_path, cdn_key, {:file, source}, public?(repository)}
 
         case upload_type do
           :both -> [versioned, unversioned]
@@ -210,8 +227,12 @@ defmodule Hexdocs.Bucket do
       {bucket(public?), store_key, data, opts}
     end)
     |> Task.async_stream(
-      fn {bucket, key, data, opts} ->
-        put(bucket, key, data, opts)
+      fn
+        {bucket, key, {:file, source}, opts} ->
+          put_file(bucket, key, source, opts)
+
+        {bucket, key, data, opts} ->
+          put(bucket, key, data, opts)
       end,
       max_concurrency: 10,
       timeout: 60_000
@@ -239,10 +260,6 @@ defmodule Hexdocs.Bucket do
         &delete_key?(&1, paths, repository, package, versions, upload_type)
       )
 
-    Enum.each(keys_to_delete, fn key ->
-      Logger.info("Deleting #{bucket} #{key}")
-    end)
-
     Hexdocs.Store.delete_many(bucket, keys_to_delete)
   end
 
@@ -332,7 +349,10 @@ defmodule Hexdocs.Bucket do
   end
 
   defp put(bucket, key, data, opts) do
-    Logger.info("Uploading #{bucket} #{key}")
     Hexdocs.Store.put!(bucket, key, data, opts)
   end
+
+  defp put_file(bucket, key, source, opts) do
+    Hexdocs.Store.put_file!(bucket, key, source, opts)
+  end
 end

lib/hexdocs/http.ex

@@ -70,6 +70,24 @@ defmodule Hexdocs.HTTP do
     end
   end
 
+  def put_file(url, headers, path) do
+    body = File.stream!(path, 65_536)
+
+    case Req.put(url,
+           headers: headers,
+           body: body,
+           retry: false,
+           decode_body: false,
+           receive_timeout: @receive_timeout
+         ) do
+      {:ok, response} ->
+        {:ok, response.status, normalize_headers(response.headers), response.body}
+
+      {:error, reason} ->
+        {:error, reason}
+    end
+  end
+
   def post(url, headers, body, opts \\ []) do
     timeout = Keyword.get(opts, :receive_timeout, @receive_timeout)
 

lib/hexdocs/queue.ex

@@ -58,16 +58,26 @@ defmodule Hexdocs.Queue do
 
     case key_components(key) do
       {:ok, repository, package, version} ->
-        body = Hexdocs.Store.get(:repo_bucket, key)
-
-        case Hexdocs.Tar.unpack(body, repository: repository, package: package, version: version) do
-          {:ok, files} ->
-            update_index_sitemap(repository, key)
-            update_package_sitemap(repository, key, package, files)
-            Logger.info("#{key}: done")
-
-          {:error, reason} ->
-            Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+        tarball_path = Hexdocs.TmpDir.tmp_file("docs-tarball")
+
+        case Hexdocs.Store.get_to_file(:repo_bucket, key, tarball_path) do
+          :ok ->
+            case Hexdocs.Tar.unpack_to_dir({:file, tarball_path},
+                   repository: repository,
+                   package: package,
+                   version: version
+                 ) do
+              {:ok, _dir, files} ->
+                update_index_sitemap(repository, key)
+                update_package_sitemap(repository, key, package, files)
+                Logger.info("#{key}: done")
+
+              {:error, reason} ->
+                Logger.error("Failed unpack #{repository}/#{package} #{version}: #{reason}")
+            end
+
+          nil ->
+            Logger.error("#{key}: package not found in store")
         end
 
       :error ->
@@ -104,26 +114,28 @@ defmodule Hexdocs.Queue do
           version: version
         })
 
-        body = Hexdocs.Store.get(:repo_bucket, key)
+        tarball_path = Hexdocs.TmpDir.tmp_file("docs-tarball")
 
-        if body do
-          case type do
-            :upload ->
-              process_upload(key, repository, package, version, body, start)
+        case Hexdocs.Store.get_to_file(:repo_bucket, key, tarball_path) do
+          :ok ->
+            case type do
+              :upload ->
+                process_upload(key, repository, package, version, {:file, tarball_path}, start)
 
-            :search ->
-              process_search(key, repository, package, version, body, start)
-          end
-        else
-          Logger.error("#{log_prefix} #{key}: package not found in store")
+              :search ->
+                process_search(key, repository, package, version, {:file, tarball_path}, start)
+            end
+
+          nil ->
+            Logger.error("#{log_prefix} #{key}: package not found in store")
         end
 
       :error ->
         Logger.info("#{key}: skip")
     end
   end
 
-  defp process_upload(key, repository, package, version, body, start) do
+  defp process_upload(key, repository, package, version, input, start) do
     {version, all_versions} =
       if package in @special_package_names do
         version =
@@ -144,15 +156,20 @@ defmodule Hexdocs.Queue do
         {version, all_versions}
       end
 
-    case Hexdocs.Tar.unpack(body, repository: repository, package: package, version: version) do
-      {:ok, files} ->
-        files = rewrite_files(files)
+    case Hexdocs.Tar.unpack_to_dir(input,
+           repository: repository,
+           package: package,
+           version: version
+         ) do
+      {:ok, dir, files} ->
+        rewrite_files(dir, files)
 
         Hexdocs.Bucket.upload(
           repository,
           package,
           version,
           all_versions,
+          dir,
           files
         )
 
@@ -170,7 +187,7 @@ defmodule Hexdocs.Queue do
     end
   end
 
-  defp process_search(key, repository, package, version, body, start) do
+  defp process_search(key, repository, package, version, input, start) do
     if repository != "hexpm" do
       Logger.warning("SKIPPING SEARCH INDEX #{key} (repository is not hexpm)")
     else
@@ -180,9 +197,9 @@ defmodule Hexdocs.Queue do
           :error when package in @special_package_names -> version
         end
 
-      case Hexdocs.Tar.unpack(body, package: package, version: version) do
-        {:ok, files} ->
-          update_search_index(key, package, version, files)
+      case Hexdocs.Tar.unpack_to_dir(input, package: package, version: version) do
+        {:ok, dir, files} ->
+          update_search_index(key, package, version, dir, files)
           elapsed = System.os_time(:millisecond) - start
           Logger.info("FINISHED INDEXING DOCS #{key} #{elapsed}ms")
 
@@ -279,9 +296,12 @@ defmodule Hexdocs.Queue do
     {package, version}
   end
 
-  defp rewrite_files(files) do
-    Enum.map(files, fn {path, content} ->
-      {path, Hexdocs.FileRewriter.run(path, content)}
+  defp rewrite_files(dir, files) do
+    Enum.each(files, fn path ->
+      full_path = Path.join(dir, path)
+      content = File.read!(full_path)
+      rewritten = Hexdocs.FileRewriter.run(path, content)
+      File.write!(full_path, rewritten)
     end)
   end
 
@@ -314,7 +334,7 @@ defmodule Hexdocs.Queue do
   defp update_package_sitemap("hexpm", key, package, files) do
     Logger.info("UPDATING PACKAGE SITEMAP #{key}")
 
-    pages = for {path, _content} <- files, Path.extname(path) == ".html", do: path
+    pages = for path <- files, Path.extname(path) == ".html", do: path
     body = Hexdocs.PackageSitemap.render(package, pages, DateTime.utc_now())
     Hexdocs.Bucket.upload_package_sitemap(package, body)
 
@@ -344,8 +364,13 @@ defmodule Hexdocs.Queue do
     :ok
   end
 
-  defp update_search_index(key, package, version, files) do
-    case Hexdocs.Search.find_search_items(package, version, files) do
+  defp update_search_index(key, package, version, dir, files) do
+    files_with_content =
+      Enum.map(files, fn path ->
+        {path, File.read!(Path.join(dir, path))}
+      end)
+
+    case Hexdocs.Search.find_search_items(package, version, files_with_content) do
       {proglang, items} ->
         Logger.info("DELETING SEARCH INDEX #{key}")
         Hexdocs.Search.delete(package, version)

lib/hexdocs/store/gs.ex

@@ -52,6 +52,18 @@ defmodule Hexdocs.Store.GS do
   end
 
   def put!(bucket, key, blob, opts) do
+    upload(bucket, key, opts, fn url, headers ->
+      Hexdocs.HTTP.put(url, headers, blob)
+    end)
+  end
+
+  def put_file!(bucket, key, source, opts) do
+    upload(bucket, key, opts, fn url, headers ->
+      Hexdocs.HTTP.put_file(url, headers, source)
+    end)
+  end
+
+  defp upload(bucket, key, opts, fun) do
     headers =
       headers() ++
         meta_headers(Keyword.fetch!(opts, :meta)) ++
@@ -64,7 +76,7 @@ defmodule Hexdocs.Store.GS do
     headers = filter_nil_values(headers)
 
     {:ok, 200, _headers, _body} =
-      Hexdocs.HTTP.retry("gs", url, fn -> Hexdocs.HTTP.put(url, headers, blob) end)
+      Hexdocs.HTTP.retry("gs", url, fn -> fun.(url, headers) end)
 
     :ok
   end