Upload S3 multipart : envoi d’erreurs sur Sentry, tweak timeout et concurrency pour la consolidation IRVE (#5551)

Co-authored-by: Thibaut Barrère <thibaut.barrere@gmail.com>

Author: vdegove

apps/shared/lib/s3.ex

@@ -35,15 +35,34 @@ defmodule Transport.S3 do
     bucket |> ExAws.S3.delete_object(path) |> Transport.Wrapper.ExAWS.impl().request!()
   end
 
-  @spec stream_to_s3!(bucket_feature(), binary(), binary(), acl: atom(), cache_control: binary()) :: any()
+  @spec stream_to_s3!(bucket_feature(), binary(), binary(),
+          acl: atom(),
+          cache_control: binary(),
+          timeout: pos_integer(),
+          max_concurrency: pos_integer()
+        ) :: any()
   def stream_to_s3!(feature, local_path, upload_path, options \\ []) do
     Logger.debug("Streaming #{local_path} to #{upload_path}")
-    options = Keyword.validate!(options, [:cache_control, {:acl, :private}])
+    options = Keyword.validate!(options, [:cache_control, :timeout, :max_concurrency, {:acl, :private}])
 
     local_path
     |> ExAws.S3.Upload.stream_file()
     |> ExAws.S3.upload(Transport.S3.bucket_name(feature), upload_path, options)
     |> Transport.Wrapper.ExAWS.impl().request!()
+  catch
+    # ExAws uploads multipart parts via `Task.async_stream`; a large/slow upload can hit its
+    # timeout, which surfaces as a process *exit* (not an exception, so it escapes `rescue` and
+    # `Sentry.capture_exception`). Report it for observability, then let it propagate unchanged.
+    # NOTE: relies on current internal behaviour of ExAws here:
+    # https://github.com/ex-aws/ex_aws_s3/blob/v2.5.9/lib/ex_aws/s3/upload.ex#L133-L141
+    :exit, reason ->
+      Sentry.capture_message("S3 upload failed (exit), likely a timeout",
+        level: :error,
+        result: :sync,
+        extra: %{feature: inspect(feature), upload_path: upload_path, reason: inspect(reason)}
+      )
+
+      exit(reason)
   end
 
   @spec download_file(bucket_feature(), binary(), binary()) :: any()

apps/shared/test/support/s3_test_utils.ex

@@ -43,9 +43,12 @@ defmodule Transport.Test.S3TestUtils do
                               src: src = %File.Stream{},
                               bucket: ^expected_bucket,
                               path: path,
-                              opts: [acl: ^expected_acl],
+                              opts: opts,
                               service: :s3
                             } ->
+      # match `acl` within the opts rather than the whole list, so callers may pass extra
+      # options (e.g. `timeout:`) without breaking this expectation
+      assert Keyword.get(opts, :acl) == expected_acl
       assert String.starts_with?(path, expected_start_path)
       assert src |> Enum.join("\n") == expected_file_content
     end)

apps/transport/lib/S3/aggregates_uploader.ex

@@ -3,24 +3,28 @@ defmodule Transport.S3.AggregatesUploader do
   Helpers to upload a file, computes its sha256, and update a "latest" file.
   """
 
-  @spec upload_aggregate!(Path.t(), String.t(), String.t()) :: :ok
+  @spec upload_aggregate!(Path.t(), String.t(), String.t(), Keyword.t()) :: :ok
   @doc """
   This method takes a local `file` and upload 4 different files to our S3 `aggregates` bucket (the bucket is expected to exist):
   - the `remote_path` and `remote_latest_path` containing the data from `file`
   - two companions files with `.sha256sum` extension appended (SHA256 sum is computed on the fly)
 
+  `options` are forwarded to `Transport.S3.stream_to_s3!/4`. Useful options:
+  - `timeout:` it applies to each part of the multipart upload (and not the whole file!). Defaults to 30s, but may be increased for large files.
+  - `max_concurrency:` it applies to the number of concurrent parts uploaded. Defaults to 4.
+
   Example
 
     with_tmp_file(fn file ->
       File.write(file, "some relevant data")
       upload_aggregate!(file, "aggregate-20250127193035.csv", "aggregate-latest.csv")
     end)
   """
-  def upload_aggregate!(file, remote_path, remote_latest_path) do
+  def upload_aggregate!(file, remote_path, remote_latest_path, options \\ []) do
     with_tmp_file(fn checksum_file ->
       sha256!(file, checksum_file)
 
-      upload_files!(file, checksum_file, remote_path)
+      upload_files!(file, checksum_file, remote_path, options)
       |> update_latest_files!(remote_latest_path)
     end)
   end
@@ -61,11 +65,11 @@ defmodule Transport.S3.AggregatesUploader do
     File.write!(checksum_file, hash)
   end
 
-  defp upload_files!(file, checksum_file, remote_path) do
+  defp upload_files!(file, checksum_file, remote_path, options) do
     remote_checksum_path = checksum_filename(remote_path)
 
-    stream_upload!(file, remote_path)
-    stream_upload!(checksum_file, remote_checksum_path)
+    stream_upload!(file, remote_path, options)
+    stream_upload!(checksum_file, remote_checksum_path, options)
 
     {remote_path, remote_checksum_path}
   end
@@ -83,8 +87,8 @@ defmodule Transport.S3.AggregatesUploader do
     "#{base_filename}.sha256sum"
   end
 
-  defp stream_upload!(file, filename) do
-    Transport.S3.stream_to_s3!(:aggregates, file, filename)
+  defp stream_upload!(file, filename, options) do
+    Transport.S3.stream_to_s3!(:aggregates, file, filename, options)
   end
 
   defp copy!(s3_path, filename) do

apps/transport/lib/irve/consolidation.ex

@@ -19,6 +19,12 @@ defmodule Transport.IRVE.Consolidation do
   @consolidated_file_no_dedup_base_name "consolidation_transport_avec_doublons_irve_statique"
   @consolidated_file_base_name "consolidation_transport_irve_statique"
 
+  # The consolidated files can be large; uploading them can exceed ExAws' default 30s multipart
+  # timeout. We lower the concurrency of the per-file parts (each part then gets more bandwidth
+  # and uploads faster) and give each part a 1 minute ceiling.
+  @s3_upload_timeout :timer.minutes(1)
+  @s3_upload_max_concurrency 2
+
   def process(opts \\ []) do
     destination = Keyword.get(opts, :destination, :send_to_s3)
     debug = Keyword.get(opts, :debug, false)
@@ -164,7 +170,9 @@ defmodule Transport.IRVE.Consolidation do
           upload_aggregate!(
             report_file,
             "#{@report_output_base_name}_#{timestamp()}.csv",
-            "#{@report_output_base_name}.csv"
+            "#{@report_output_base_name}.csv",
+            timeout: @s3_upload_timeout,
+            max_concurrency: @s3_upload_max_concurrency
           )
         end)
 
@@ -186,7 +194,9 @@ defmodule Transport.IRVE.Consolidation do
       upload_aggregate!(
         consolidation_file,
         "#{base_name}_#{timestamp()}.csv",
-        "#{base_name}.csv"
+        "#{base_name}.csv",
+        timeout: @s3_upload_timeout,
+        max_concurrency: @s3_upload_max_concurrency
       )
     end)
   end