diff --git a/lib/waffle/http_client/content_disposition.ex b/lib/waffle/http_client/content_disposition.ex
new file mode 100644
index 0000000..82a88c0
--- /dev/null
+++ b/lib/waffle/http_client/content_disposition.ex
@@ -0,0 +1,121 @@
+defmodule Waffle.HTTPClient.ContentDisposition do
+  @moduledoc """
+  Parses filenames out of `Content-Disposition` header values.
+
+  Supports both the RFC 5987 extended notation (`filename*=...`), which
+  takes priority per RFC 6266 when both are present, and the plain
+  `filename=` parameter (quoted or unquoted).
+  """
+
+  @doc """
+  Extracts the filename from a `Content-Disposition` header value.
+
+  Returns `nil` if no filename parameter is present.
+
+  ## Examples
+
+      iex> Waffle.HTTPClient.ContentDisposition.filename(~s(attachment; filename="photo.jpg"))
+      "photo.jpg"
+
+      iex> Waffle.HTTPClient.ContentDisposition.filename("attachment; filename*=UTF-8''my%20photo.jpg")
+      "my photo.jpg"
+
+      iex> Waffle.HTTPClient.ContentDisposition.filename("inline")
+      nil
+
+  """
+  @spec filename(String.t()) :: String.t() | nil
+  def filename(value) when is_binary(value) do
+    params = parse_params(value)
+
+    case Map.fetch(params, "filename*") do
+      {:ok, extended_value} ->
+        decode_extended_value(extended_value)
+
+      :error ->
+        Map.get(params, "filename")
+    end
+  end
+
+  # Splits "type; key=value; key2="value 2"" into a map of downcased
+  # parameter names to their unquoted/unescaped values. The leading
+  # disposition-type token (e.g. "attachment"/"inline") is discarded.
+  defp parse_params(value) do
+    value
+    |> split_semicolons()
+    |> Enum.drop(1)
+    |> Enum.reduce(%{}, fn part, acc ->
+      case parse_param(String.trim(part)) do
+        {key, val} -> Map.put_new(acc, key, val)
+        :error -> acc
+      end
+    end)
+  end
+
+  # Splits on `;`, while treating the contents of a double-quoted string as
+  # opaque so that a `;` inside a quoted filename doesn't split it.
+  defp split_semicolons(value), do: split_semicolons(value, <<>>, [], false)
+
+  defp split_semicolons(<<?\\, char, rest::binary>>, acc, parts, true) do
+    split_semicolons(rest, <<acc::binary, ?\\, char>>, parts, true)
+  end
+
+  defp split_semicolons(<<?", rest::binary>>, acc, parts, quoted?) do
+    split_semicolons(rest, <<acc::binary, ?">>, parts, not quoted?)
+  end
+
+  defp split_semicolons(<<?;, rest::binary>>, acc, parts, false) do
+    split_semicolons(rest, <<>>, [acc | parts], false)
+  end
+
+  defp split_semicolons(<<char, rest::binary>>, acc, parts, quoted?) do
+    split_semicolons(rest, <<acc::binary, char>>, parts, quoted?)
+  end
+
+  defp split_semicolons(<<>>, acc, parts, _quoted?) do
+    Enum.reverse([acc | parts])
+  end
+
+  # Parses a single "key=value" parameter, downcasing the key and
+  # unquoting/unescaping the value if it's a quoted string. Returns
+  # `:error` for anything that isn't a well-formed, non-empty parameter.
+  defp parse_param(param) do
+    with [key, value] <- :binary.split(param, "="),
+         value <- value |> String.trim() |> unquote_value(),
+         false <- value == "" do
+      {key |> String.trim() |> String.downcase(), value}
+    else
+      _ -> :error
+    end
+  end
+
+  defp unquote_value(<<?", rest::binary>>), do: parse_quoted(rest, <<>>)
+  defp unquote_value(value), do: unquoted_token(value, <<>>)
+
+  defp parse_quoted(<<?\\, char, rest::binary>>, acc), do: parse_quoted(rest, <<acc::binary, char>>)
+  defp parse_quoted(<<?", _rest::binary>>, acc), do: acc
+  defp parse_quoted(<<char, rest::binary>>, acc), do: parse_quoted(rest, <<acc::binary, char>>)
+  defp parse_quoted(<<>>, acc), do: acc
+
+  defp unquoted_token(<<>>, acc), do: acc
+  defp unquoted_token(<<char, _rest::binary>>, acc) when char in [?\s, ?\t], do: acc
+  defp unquoted_token(<<char, rest::binary>>, acc), do: unquoted_token(rest, <<acc::binary, char>>)
+
+  # RFC 5987: charset'language'percent-encoded-value,
+  # e.g. UTF-8''my%20photo.jpg or UTF-8'en'my%20photo.jpg
+  defp decode_extended_value(value) do
+    case String.split(value, "'", parts: 3) do
+      [_charset, _language, encoded] -> safe_uri_decode(encoded)
+      _ -> safe_uri_decode(value)
+    end
+  end
+
+  # A malformed percent-encoding (e.g. a trailing "%" or "%zz") makes
+  # `URI.decode/1` raise. Fall back to the raw (un-decoded) value rather than
+  # crashing the caller over a malformed header from a remote server.
+  defp safe_uri_decode(value) do
+    URI.decode(value)
+  rescue
+    ArgumentError -> value
+  end
+end
diff --git a/lib/waffle/http_client/hackney.ex b/lib/waffle/http_client/hackney.ex
index 2df6964..f2a52c8 100644
--- a/lib/waffle/http_client/hackney.ex
+++ b/lib/waffle/http_client/hackney.ex
@@ -4,7 +4,7 @@ defmodule Waffle.HTTPClient.Hackney do
 
   Add `:hackney` to your dependencies:
 
-      {:hackney, "~> 1.9"}
+      {:hackney, "~> 4.5.2"}
 
   ## Configuration
 
@@ -18,54 +18,211 @@ defmodule Waffle.HTTPClient.Hackney do
   | `:connect_timeout` | `10_000`     | Timeout for establishing a connection, in milliseconds |
   | `:max_body_length` | `:infinity`  | Maximum response body size, in bytes                   |
   | `:follow_redirect` | `true`       | Whether to follow HTTP redirects automatically         |
+  | `:max_redirect`    | `5`          | Maximum number of redirects to follow                  |
   """
 
   @behaviour Waffle.HTTPClient
 
   @impl Waffle.HTTPClient
   def get(url, headers, options) do
-    hackney_options = [
-      follow_redirect: Keyword.get(options, :follow_redirect, true),
-      recv_timeout: Keyword.get(options, :recv_timeout, 5_000),
-      connect_timeout: Keyword.get(options, :connect_timeout, 10_000)
-    ]
+    do_get(url, headers, options, 0)
+  end
 
+  defp do_get(url, headers, options, redirect_count) do
+    connect_timeout = Keyword.get(options, :connect_timeout, 10_000)
+    recv_timeout = Keyword.get(options, :recv_timeout, 5_000)
+    follow_redirect = Keyword.get(options, :follow_redirect, true)
+    max_redirect = Keyword.get(options, :max_redirect, 5)
     max_body_length = Keyword.get(options, :max_body_length, :infinity)
 
+    hackney_options = [
+      {:async, :once},
+      follow_redirect: follow_redirect,
+      recv_timeout: recv_timeout,
+      connect_timeout: connect_timeout,
+      # Force HTTP/1.1: hackney 4.5.2's HTTP/2 async response dispatch
+      # (hackney_conn:h2_on_response/4, h2_on_data/4) never delivers
+      # `hackney_response` messages to the caller.
+      # See https://github.com/benoitc/hackney/issues/909
+      protocols: [:http1]
+    ]
+
     case :hackney.get(url, headers, "", hackney_options) do
-      {:ok, 200, response_headers, client_ref} ->
-        read_body(client_ref, response_headers, max_body_length)
+      {:ok, ref} ->
+        receive_status(
+          ref,
+          max_body_length,
+          recv_timeout,
+          url,
+          headers,
+          options,
+          max_redirect,
+          redirect_count
+        )
+
+      {:error, reason} ->
+        normalize_error(reason)
+    end
+  end
 
-      {:ok, 503, _headers, client_ref} ->
-        :hackney.close(client_ref)
+  defp receive_status(
+         ref,
+         max_body_length,
+         recv_timeout,
+         url,
+         headers,
+         options,
+         max_redirect,
+         redirect_count
+       ) do
+    receive do
+      {:hackney_response, ^ref, {:status, 200, _reason}} ->
+        receive_headers(ref, max_body_length, recv_timeout)
+
+      {:hackney_response, ^ref, {:status, 503, _reason}} ->
+        close_and_flush(ref)
         {:error, :service_unavailable}
 
-      {:ok, status, _headers, client_ref} ->
-        :hackney.close(client_ref)
+      {:hackney_response, ^ref, {:redirect, location, _response_headers}} ->
+        follow_redirect(ref, url, location, headers, options, max_redirect, redirect_count)
+
+      {:hackney_response, ^ref, {:see_other, location, _response_headers}} ->
+        follow_redirect(ref, url, location, headers, options, max_redirect, redirect_count)
+
+      {:hackney_response, ^ref, {:status, status, _reason}} ->
+        close_and_flush(ref)
         {:error, {:http_error, status}}
 
-      {:error, reason} ->
+      {:hackney_response, ^ref, {:error, reason}} ->
+        close_and_flush(ref)
         normalize_error(reason)
+    after
+      recv_timeout ->
+        close_and_flush(ref)
+        {:error, :recv_timeout}
     end
   end
 
-  defp read_body(client_ref, response_headers, max_body_length) do
-    case :hackney.body(client_ref, max_body_length) do
-      {:ok, body} ->
-        filename =
-          :hackney_headers.new(response_headers)
-          |> get_content_disposition_filename()
+  defp follow_redirect(ref, url, location, headers, options, max_redirect, redirect_count) do
+    close_and_flush(ref)
 
-        if filename, do: {:ok, body, filename}, else: {:ok, body}
+    if redirect_count >= max_redirect do
+      {:error, {:too_many_redirects, redirect_count}}
+    else
+      new_url = resolve_redirect_url(url, location)
+      new_headers = strip_sensitive_headers_on_cross_origin(url, new_url, headers)
+      do_get(new_url, new_headers, options, redirect_count + 1)
+    end
+  end
 
-      {:error, reason} ->
-        :hackney.close(client_ref)
+  defp resolve_redirect_url(url, location) do
+    location = to_string(location)
+
+    url
+    |> URI.merge(location)
+    |> URI.to_string()
+  end
+
+  # Mirrors hackney's own CVE-2018-1000007 mitigation for its sync client:
+  # don't forward credentials to a different host on redirect.
+  defp strip_sensitive_headers_on_cross_origin(url, new_url, headers) do
+    if same_origin?(url, new_url) do
+      headers
+    else
+      Enum.reject(headers, fn {key, _value} ->
+        String.downcase(to_string(key)) in ["authorization", "cookie"]
+      end)
+    end
+  end
+
+  defp same_origin?(url, new_url) do
+    a = URI.parse(url)
+    b = URI.parse(new_url)
+    {a.scheme, a.host, a.port} == {b.scheme, b.host, b.port}
+  end
+
+  defp receive_headers(ref, max_body_length, recv_timeout) do
+    receive do
+      {:hackney_response, ^ref, {:headers, response_headers}} ->
+        :hackney.stream_next(ref)
+        receive_body(ref, response_headers, [], 0, max_body_length, recv_timeout)
+
+      {:hackney_response, ^ref, {:error, reason}} ->
+        close_and_flush(ref)
         normalize_error(reason)
+    after
+      recv_timeout ->
+        close_and_flush(ref)
+        {:error, :recv_timeout}
     end
   end
 
-  # connect timeout: hackney returns %{reason: :timeout}, not a bare atom
+  defp receive_body(ref, response_headers, acc, size, max_body_length, recv_timeout) do
+    receive do
+      {:hackney_response, ^ref, :done} ->
+        body = acc |> Enum.reverse() |> IO.iodata_to_binary()
+        build_response(body, response_headers)
+
+      {:hackney_response, ^ref, chunk} when is_binary(chunk) ->
+        new_size = size + byte_size(chunk)
+
+        if max_body_length != :infinity and new_size > max_body_length do
+          close_and_flush(ref)
+          {:error, {:http_error, :body_too_large}}
+        else
+          :hackney.stream_next(ref)
+
+          receive_body(
+            ref,
+            response_headers,
+            [chunk | acc],
+            new_size,
+            max_body_length,
+            recv_timeout
+          )
+        end
+
+      {:hackney_response, ^ref, {:error, reason}} ->
+        close_and_flush(ref)
+        normalize_error(reason)
+    after
+      recv_timeout ->
+        close_and_flush(ref)
+        {:error, :recv_timeout}
+    end
+  end
+
+  defp build_response(body, response_headers) do
+    filename =
+      :hackney_headers.new(response_headers)
+      |> get_content_disposition_filename()
+
+    if filename, do: {:ok, body, filename}, else: {:ok, body}
+  end
+
+  # Closes the connection and drains any `{:hackney_response, ^ref, _}`
+  # messages already sitting in our mailbox. Hackney can push multiple
+  # messages (e.g. status + headers, or a trailing chunk) before we act on
+  # an error/abort, and without this a long-lived process (e.g. a GenServer
+  # calling into Waffle) would accumulate stray `handle_info` messages.
+  defp close_and_flush(ref) do
+    :hackney.close(ref)
+    flush_messages(ref)
+  end
+
+  defp flush_messages(ref) do
+    receive do
+      {:hackney_response, ^ref, _msg} -> flush_messages(ref)
+    after
+      0 -> :ok
+    end
+  end
+
+  # connect/checkout timeout: hackney returns %{reason: :timeout} (older
+  # shape) or the bare atoms :connect_timeout / :checkout_timeout (4.5.2)
   defp normalize_error(%{reason: :timeout}), do: {:error, :timeout}
+  defp normalize_error(:connect_timeout), do: {:error, :timeout}
+  defp normalize_error(:checkout_timeout), do: {:error, :timeout}
   # recv timeout: hackney returns a bare :timeout atom
   defp normalize_error(:timeout), do: {:error, :recv_timeout}
   defp normalize_error(reason), do: {:error, {:http_error, reason}}
@@ -75,11 +232,11 @@ defmodule Waffle.HTTPClient.Hackney do
       :undefined ->
         nil
 
-      value ->
-        case :hackney_headers.content_disposition(value) do
-          {_, [{"filename", filename} | _]} -> filename
-          _ -> nil
-        end
+      value when is_binary(value) ->
+        Waffle.HTTPClient.ContentDisposition.filename(value)
+
+      _ ->
+        nil
     end
   end
 end
diff --git a/mix.exs b/mix.exs
index 66735ab..894ca4f 100644
--- a/mix.exs
+++ b/mix.exs
@@ -11,6 +11,7 @@ defmodule Waffle.Mixfile do
       source_url: "https://github.com/elixir-waffle/waffle",
       deps: deps(),
       docs: docs(),
+      elixirc_paths: elixirc_paths(Mix.env()),
 
       # Hex
       description: description(),
@@ -18,6 +19,9 @@ defmodule Waffle.Mixfile do
     ]
   end
 
+  defp elixirc_paths(:test), do: ["lib", "test/support"]
+  defp elixirc_paths(_), do: ["lib"]
+
   defp description do
     """
     Flexible file upload and attachment library for Elixir.
@@ -58,7 +62,7 @@ defmodule Waffle.Mixfile do
 
   defp deps do
     [
-      {:hackney, "~> 1.9"},
+      {:hackney, "~> 4.5.2"},
 
       # If using Amazon S3
       {:ex_aws, "~> 2.1", optional: true},
diff --git a/mix.lock b/mix.lock
index 6380dde..e1ee0b7 100644
--- a/mix.lock
+++ b/mix.lock
@@ -1,28 +1,31 @@
 %{
   "bunt": {:hex, :bunt, "1.0.0", "081c2c665f086849e6d57900292b3a161727ab40431219529f13c4ddcf3e7a44", [:mix], [], "hexpm", "dc5f86aa08a5f6fa6b8096f0735c4e76d54ae5c9fa2c143e5a1fc7c1cd9bb6b5"},
-  "certifi": {:hex, :certifi, "2.9.0", "6f2a475689dd47f19fb74334859d460a2dc4e3252a3324bd2111b8f0429e7e21", [:rebar3], [], "hexpm", "266da46bdb06d6c6d35fde799bcb28d36d985d424ad7c08b5bb48f5b5cdd4641"},
+  "certifi": {:hex, :certifi, "2.17.0", "835748414307e15e05b17d0e518190228ce648b08d569a5cc93a85a40f3e5c9b", [:rebar3], [], "hexpm", "8122798a17f0293c80daada25d0f81c7f4d708c73fef782c7c9b1950e26e4d21"},
   "credo": {:hex, :credo, "1.7.19", "cc52129665fc7c15143d47838fda0f9cd6dac9ceced7bf4da6f85fcbfe64b12a", [:mix], [{:bunt, "~> 0.2.1 or ~> 1.0", [hex: :bunt, repo: "hexpm", optional: false]}, {:file_system, "~> 0.2 or ~> 1.0", [hex: :file_system, repo: "hexpm", optional: false]}, {:jason, "~> 1.0", [hex: :jason, repo: "hexpm", optional: false]}], "hexpm", "2d8bc95d5a7bb99dd2613621d4f08c6a3575c3fd4b62e6a2b48a100352a557b8"},
   "earmark": {:hex, :earmark, "1.4.4", "4821b8d05cda507189d51f2caeef370cf1e18ca5d7dfb7d31e9cafe6688106a4", [:mix], [], "hexpm"},
   "earmark_parser": {:hex, :earmark_parser, "1.4.39", "424642f8335b05bb9eb611aa1564c148a8ee35c9c8a8bba6e129d51a3e3c6769", [:mix], [], "hexpm", "06553a88d1f1846da9ef066b87b57c6f605552cfbe40d20bd8d59cc6bde41944"},
-  "ex_aws": {:hex, :ex_aws, "2.4.1", "d1dc8965d1dc1c939dd4570e37f9f1d21e047e4ecd4f9373dc89cd4e45dce5ef", [:mix], [{:configparser_ex, "~> 4.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 1.16", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8 or ~> 3.0", [hex: :jsx, repo: "hexpm", optional: true]}, {:mime, "~> 1.2 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:sweet_xml, "~> 0.7", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "803387db51b4e91be4bf0110ba999003ec6103de7028b808ee9b01f28dbb9eee"},
+  "ex_aws": {:hex, :ex_aws, "2.7.0", "e6bfd4b5fb8c791aa6c7d57fc7c45f050bb89de9576f1f6104aa974b234c01ec", [:mix], [{:configparser_ex, "~> 5.0", [hex: :configparser_ex, repo: "hexpm", optional: true]}, {:hackney, "~> 4.0", [hex: :hackney, repo: "hexpm", optional: true]}, {:jason, "~> 1.1", [hex: :jason, repo: "hexpm", optional: true]}, {:jsx, "~> 2.8 or ~> 3.0", [hex: :jsx, repo: "hexpm", optional: true]}, {:mime, "~> 1.2 or ~> 2.0", [hex: :mime, repo: "hexpm", optional: false]}, {:req, "~> 0.5.10 or ~> 0.6 or ~> 1.0", [hex: :req, repo: "hexpm", optional: true]}, {:sweet_xml, "~> 0.7", [hex: :sweet_xml, repo: "hexpm", optional: true]}, {:telemetry, "~> 0.4.3 or ~> 1.0", [hex: :telemetry, repo: "hexpm", optional: false]}], "hexpm", "bfe9d744d4fd4c1f40314ee7fab504d5547d1f01cd377fff1568cbe630b06d65"},
   "ex_aws_s3": {:hex, :ex_aws_s3, "2.4.0", "ce8decb6b523381812798396bc0e3aaa62282e1b40520125d1f4eff4abdff0f4", [:mix], [{:ex_aws, "~> 2.0", [hex: :ex_aws, repo: "hexpm", optional: false]}, {:sweet_xml, ">= 0.0.0", [hex: :sweet_xml, repo: "hexpm", optional: true]}], "hexpm", "85dda6e27754d94582869d39cba3241d9ea60b6aa4167f9c88e309dc687e56bb"},
   "ex_doc": {:hex, :ex_doc, "0.34.0", "ab95e0775db3df71d30cf8d78728dd9261c355c81382bcd4cefdc74610bef13e", [:mix], [{:earmark_parser, "~> 1.4.39", [hex: :earmark_parser, repo: "hexpm", optional: false]}, {:makeup_c, ">= 0.1.0", [hex: :makeup_c, repo: "hexpm", optional: true]}, {:makeup_elixir, "~> 0.14 or ~> 1.0", [hex: :makeup_elixir, repo: "hexpm", optional: false]}, {:makeup_erlang, "~> 0.1 or ~> 1.0", [hex: :makeup_erlang, repo: "hexpm", optional: false]}, {:makeup_html, ">= 0.1.0", [hex: :makeup_html, repo: "hexpm", optional: true]}], "hexpm", "60734fb4c1353f270c3286df4a0d51e65a2c1d9fba66af3940847cc65a8066d7"},
   "file_system": {:hex, :file_system, "1.1.1", "31864f4685b0148f25bd3fbef2b1228457c0c89024ad67f7a81a3ffbc0bbad3a", [:mix], [], "hexpm", "7a15ff97dfe526aeefb090a7a9d3d03aa907e100e262a0f8f7746b78f8f87a5d"},
-  "hackney": {:hex, :hackney, "1.18.1", "f48bf88f521f2a229fc7bae88cf4f85adc9cd9bcf23b5dc8eb6a1788c662c4f6", [:rebar3], [{:certifi, "~> 2.9.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:idna, "~> 6.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:metrics, "~> 1.0.0", [hex: :metrics, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.1", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.3.1", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "a4ecdaff44297e9b5894ae499e9a070ea1888c84afdd1fd9b7b2bc384950128e"},
-  "idna": {:hex, :idna, "6.1.1", "8a63070e9f7d0c62eb9d9fcb360a7de382448200fbbd1b106cc96d3d8099df8d", [:rebar3], [{:unicode_util_compat, "~> 0.7.0", [hex: :unicode_util_compat, repo: "hexpm", optional: false]}], "hexpm", "92376eb7894412ed19ac475e4a86f7b413c1b9fbb5bd16dccd57934157944cea"},
+  "h2": {:hex, :h2, "0.10.4", "31ef580056e19977b6542eea2718aa19a45841205cd82cd48535dd7683caea4c", [:rebar3], [], "hexpm", "8350920ac7c259fb759a2830d3ce0dd248c48422a031d1760bebdd2020fa30a9"},
+  "hackney": {:hex, :hackney, "4.5.2", "d44b00b6e91ef8c480ec51ad42e9d8777a3a391ea78865cb16c5c0645399c6f5", [:rebar3], [{:certifi, "~> 2.17.0", [hex: :certifi, repo: "hexpm", optional: false]}, {:h2, "~> 0.10.4", [hex: :h2, repo: "hexpm", optional: false]}, {:idna, "~> 7.1.0", [hex: :idna, repo: "hexpm", optional: false]}, {:mimerl, "~> 1.4", [hex: :mimerl, repo: "hexpm", optional: false]}, {:parse_trans, "3.4.2", [hex: :parse_trans, repo: "hexpm", optional: false]}, {:quic, "~> 1.7.0", [hex: :quic, repo: "hexpm", optional: false]}, {:ssl_verify_fun, "~> 1.1.0", [hex: :ssl_verify_fun, repo: "hexpm", optional: false]}, {:webtransport, "~> 0.4.3", [hex: :webtransport, repo: "hexpm", optional: false]}], "hexpm", "5db2e2d7de05cafcd681e21120ae4c8b5a378ebeb9b8d8835f26f456a9a80f7a"},
+  "idna": {:hex, :idna, "7.1.0", "1067a13043538129602d2f2ce6899d8713125c7d19734aa557ce2e3ea55bd4f1", [:rebar3], [], "hexpm", "6ae959a025bf36df61a8cab8508d9654891b5426a84c44d82deaffd6ddf8c71f"},
   "jason": {:hex, :jason, "1.4.5", "2e3a008590b0b8d7388c20293e9dcc9cf3e5d642fd2a114e4cbbb52e595d940a", [:mix], [{:decimal, "~> 1.0 or ~> 2.0 or ~> 3.0", [hex: :decimal, repo: "hexpm", optional: true]}], "hexpm", "b0c823996102bcd0239b3c2444eb00409b72f6a140c1950bc8b457d836b30684"},
   "makeup": {:hex, :makeup, "1.1.2", "9ba8837913bdf757787e71c1581c21f9d2455f4dd04cfca785c70bbfff1a76a3", [:mix], [{:nimble_parsec, "~> 1.2.2 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "cce1566b81fbcbd21eca8ffe808f33b221f9eee2cbc7a1706fc3da9ff18e6cac"},
   "makeup_elixir": {:hex, :makeup_elixir, "0.16.2", "627e84b8e8bf22e60a2579dad15067c755531fea049ae26ef1020cad58fe9578", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}, {:nimble_parsec, "~> 1.2.3 or ~> 1.3", [hex: :nimble_parsec, repo: "hexpm", optional: false]}], "hexpm", "41193978704763f6bbe6cc2758b84909e62984c7752b3784bd3c218bb341706b"},
   "makeup_erlang": {:hex, :makeup_erlang, "1.0.0", "6f0eff9c9c489f26b69b61440bf1b238d95badae49adac77973cbacae87e3c2e", [:mix], [{:makeup, "~> 1.0", [hex: :makeup, repo: "hexpm", optional: false]}], "hexpm", "ea7a9307de9d1548d2a72d299058d1fd2339e3d398560a0e46c27dab4891e4d2"},
   "meck": {:hex, :meck, "1.2.0", "c3618447506589a2931be80d006ea883aff156dd56e8822f7c1bb49896e94edb", [:rebar3], [], "hexpm", "a2cfd08306ef4992db7cd2aa3521b797456abec9f562f3f0341633052dee4af9"},
   "metrics": {:hex, :metrics, "1.0.1", "25f094dea2cda98213cecc3aeff09e940299d950904393b2a29d191c346a8486", [:rebar3], [], "hexpm", "69b09adddc4f74a40716ae54d140f93beb0fb8978d8636eaded0c31b6f099f16"},
-  "mime": {:hex, :mime, "2.0.3", "3676436d3d1f7b81b5a2d2bd8405f412c677558c81b1c92be58c00562bb59095", [:mix], [], "hexpm", "27a30bf0db44d25eecba73755acf4068cbfe26a4372f9eb3e4ea3a45956bff6b"},
-  "mimerl": {:hex, :mimerl, "1.2.0", "67e2d3f571088d5cfd3e550c383094b47159f3eee8ffa08e64106cdf5e981be3", [:rebar3], [], "hexpm", "f278585650aa581986264638ebf698f8bb19df297f66ad91b18910dfc6e19323"},
+  "mime": {:hex, :mime, "2.0.7", "b8d739037be7cd402aee1ba0306edfdef982687ee7e9859bee6198c1e7e2f128", [:mix], [], "hexpm", "6171188e399ee16023ffc5b76ce445eb6d9672e2e241d2df6050f3c771e80ccd"},
+  "mimerl": {:hex, :mimerl, "1.5.0", "f35aca6f23242339b3666e0ac0702379e362b469d0aea167f6cc713547e777ed", [:rebar3], [], "hexpm", "db648ce065bae14ea84ca8b5dd123f42f49417cef693541110bf6f9e9be9ecc4"},
   "mock": {:hex, :mock, "0.3.9", "10e44ad1f5962480c5c9b9fa779c6c63de9bd31997c8e04a853ec990a9d841af", [:mix], [{:meck, "~> 0.9.2", [hex: :meck, repo: "hexpm", optional: false]}], "hexpm", "9e1b244c4ca2551bb17bb8415eed89e40ee1308e0fbaed0a4fdfe3ec8a4adbd3"},
   "nimble_parsec": {:hex, :nimble_parsec, "1.4.0", "51f9b613ea62cfa97b25ccc2c1b4216e81df970acd8e16e8d1bdc58fef21370d", [:mix], [], "hexpm", "9c565862810fb383e9838c1dd2d7d2c437b3d13b267414ba6af33e50d2d1cf28"},
-  "parse_trans": {:hex, :parse_trans, "3.3.1", "16328ab840cc09919bd10dab29e431da3af9e9e7e7e6f0089dd5a2d2820011d8", [:rebar3], [], "hexpm", "07cd9577885f56362d414e8c4c4e6bdf10d43a8767abb92d24cbe8b24c54888b"},
+  "parse_trans": {:hex, :parse_trans, "3.4.2", "c352ddc1a0d5e54f9b1654d45f9c432eef76f9cea371c55ddff769ef688fdb74", [:rebar3], [], "hexpm", "4c25347de3b7c35732d32e69ab43d1ceee0beae3f3b3ade1b59cbd3dd224d9ca"},
+  "quic": {:hex, :quic, "1.7.0", "ebcf105fb16ca4d954c3bbd2e5bcda7650074f3e3a9ca80135dbee2607e0c27e", [:rebar3], [], "hexpm", "d0d304e5fda48754a570b90b103153f780f9926c15f6eabfe1b343e4f61e32df"},
   "ssl_verify_fun": {:hex, :ssl_verify_fun, "1.1.7", "354c321cf377240c7b8716899e182ce4890c5938111a1296add3ec74cf1715df", [:make, :mix, :rebar3], [], "hexpm", "fe4c190e8f37401d30167c8c405eda19469f34577987c76dde613e838bbc67f8"},
-  "sweet_xml": {:hex, :sweet_xml, "0.7.3", "debb256781c75ff6a8c5cbf7981146312b66f044a2898f453709a53e5031b45b", [:mix], [], "hexpm", "e110c867a1b3fe74bfc7dd9893aa851f0eed5518d0d7cad76d7baafd30e4f5ba"},
-  "telemetry": {:hex, :telemetry, "1.2.1", "68fdfe8d8f05a8428483a97d7aab2f268aaff24b49e0f599faa091f1d4e7f61c", [:rebar3], [], "hexpm", "dad9ce9d8effc621708f99eac538ef1cbe05d6a874dd741de2e689c47feafed5"},
+  "sweet_xml": {:hex, :sweet_xml, "0.7.5", "803a563113981aaac202a1dbd39771562d0ad31004ddbfc9b5090bdcd5605277", [:mix], [], "hexpm", "193b28a9b12891cae351d81a0cead165ffe67df1b73fe5866d10629f4faefb12"},
+  "telemetry": {:hex, :telemetry, "1.4.2", "a0cb522801dffb1c49fe6e30561badffc7b6d0e180db1300df759faa22062855", [:rebar3], [], "hexpm", "928f6495066506077862c0d1646609eed891a4326bee3126ba54b60af61febb1"},
   "unicode_util_compat": {:hex, :unicode_util_compat, "0.7.0", "bc84380c9ab48177092f43ac89e4dfa2c6d62b40b8bd132b1059ecc7232f9a78", [:rebar3], [], "hexpm", "25eee6d67df61960cf6a794239566599b09e17e668d3700247bc498638152521"},
+  "webtransport": {:hex, :webtransport, "0.4.3", "df0c53da138cdc5f3390a31ef612caa3a041c23e18c1a0b6fbd4d72743277806", [:rebar3], [{:h2, "~> 0.10.4", [hex: :h2, repo: "hexpm", optional: false]}, {:quic, "~> 1.7.0", [hex: :quic, repo: "hexpm", optional: false]}], "hexpm", "3ae8c76696cdb56bee3caf1dd853a0a596339886468ca17c578a9bfd647dab9d"},
 }
diff --git a/test/actions/store_test.exs b/test/actions/store_test.exs
index 982edb5..6a724fe 100644
--- a/test/actions/store_test.exs
+++ b/test/actions/store_test.exs
@@ -5,6 +5,7 @@ defmodule WaffleTest.Actions.Store do
   @remote_img_with_space_image_two "https://github.com/elixir-waffle/waffle/blob/master/test/support/image%20two.png"
 
   import Mock
+  import WaffleTest.Support.HackneyMock, only: [mock_hackney_messages: 1]
 
   defmodule DummyDefinition do
     use Waffle.Actions.Store
@@ -190,13 +191,29 @@ defmodule WaffleTest.Actions.Store do
   end
 
   test "sets remote filename from content-disposition header when available" do
+    response_headers = [{"content-disposition", ~s(attachment; filename="image three.png")}]
+
+    {ref, send_auto, send_next} =
+      mock_hackney_messages([
+        {:status, 200, "OK"},
+        {:headers, response_headers},
+        "fake image data",
+        :done
+      ])
+
     with_mocks([
       {
-        :hackney_headers,
-        [:passthrough],
-        get_value: fn "content-disposition", _headers ->
-          "attachment; filename=\"image three.png\""
-        end
+        :hackney,
+        [],
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end,
+        close: fn ^ref -> :ok end
       },
       {
         Waffle.Storage.S3,
@@ -212,11 +229,22 @@ defmodule WaffleTest.Actions.Store do
   end
 
   test "sets HTTP headers for request to remote file" do
+    {ref, send_auto, send_next} =
+      mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "favicon data", :done])
+
     with_mocks([
       {
         :hackney,
-        [:passthrough],
-        []
+        [],
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end,
+        close: fn ^ref -> :ok end
       },
       {
         Waffle.Storage.S3,
diff --git a/test/http_client/content_disposition_test.exs b/test/http_client/content_disposition_test.exs
new file mode 100644
index 0000000..6bce077
--- /dev/null
+++ b/test/http_client/content_disposition_test.exs
@@ -0,0 +1,95 @@
+defmodule WaffleTest.HTTPClient.ContentDisposition do
+  use ExUnit.Case, async: true
+  doctest Waffle.HTTPClient.ContentDisposition
+
+  alias Waffle.HTTPClient.ContentDisposition
+
+  describe "filename/1" do
+    test "returns nil when there is no filename parameter" do
+      assert ContentDisposition.filename("inline") == nil
+    end
+
+    test "returns nil for an attachment with no filename parameter" do
+      assert ContentDisposition.filename("attachment") == nil
+    end
+
+    test "parses a quoted filename" do
+      assert ContentDisposition.filename(~s(attachment; filename="photo.jpg")) == "photo.jpg"
+    end
+
+    test "parses an unquoted filename" do
+      assert ContentDisposition.filename("attachment; filename=photo.jpg") == "photo.jpg"
+    end
+
+    test "parses an unquoted filename terminated by a semicolon" do
+      assert ContentDisposition.filename("attachment; filename=photo.jpg; size=1234") ==
+               "photo.jpg"
+    end
+
+    test "decodes an RFC 5987 filename*= value" do
+      assert ContentDisposition.filename("attachment; filename*=UTF-8''my%20photo.jpg") ==
+               "my photo.jpg"
+    end
+
+    test "decodes an RFC 5987 filename*= value with a language tag" do
+      assert ContentDisposition.filename("attachment; filename*=UTF-8'en'my%20photo.jpg") ==
+               "my photo.jpg"
+    end
+
+    test "prefers filename*= over filename= when both are present" do
+      value = ~s(attachment; filename="fallback.jpg"; filename*=UTF-8''preferred.jpg)
+      assert ContentDisposition.filename(value) == "preferred.jpg"
+    end
+
+    test "prefers filename*= over filename= regardless of parameter order" do
+      value = ~s(attachment; filename*=UTF-8''preferred.jpg; filename="fallback.jpg")
+      assert ContentDisposition.filename(value) == "preferred.jpg"
+    end
+
+    test "handles filename= with no value gracefully" do
+      assert ContentDisposition.filename("attachment; filename=") == nil
+    end
+
+    test "is case-insensitive for the filename parameter name" do
+      assert ContentDisposition.filename(~s(attachment; FILENAME="photo.jpg")) == "photo.jpg"
+    end
+
+    test "unescapes backslash-escaped quotes inside a quoted filename" do
+      value = ~S(attachment; filename="my \"quoted\" file.jpg")
+      assert ContentDisposition.filename(value) == ~s(my "quoted" file.jpg)
+    end
+
+    test "unescapes backslash-escaped backslashes inside a quoted filename" do
+      value = ~S(attachment; filename="C:\\Users\\file.jpg")
+      assert ContentDisposition.filename(value) == ~S(C:\Users\file.jpg)
+    end
+
+    test "does not split on a semicolon inside a quoted filename" do
+      value = ~s(attachment; filename="file; with; semicolons.jpg")
+      assert ContentDisposition.filename(value) == "file; with; semicolons.jpg"
+    end
+
+    test "handles a trailing parameter after a quoted filename" do
+      value = ~s(attachment; filename="photo.jpg"; size=1234)
+      assert ContentDisposition.filename(value) == "photo.jpg"
+    end
+
+    test "ignores whitespace around the equals sign" do
+      assert ContentDisposition.filename(~s(attachment; filename = "photo.jpg")) == "photo.jpg"
+    end
+
+    test "best-effort parses an unterminated quoted filename (missing closing quote)" do
+      assert ContentDisposition.filename(~s(attachment; filename="photo.jpg)) == "photo.jpg"
+    end
+
+    test "falls back to the raw value instead of raising on malformed percent-encoding in filename*=" do
+      value = "attachment; filename*=UTF-8''broken%"
+      assert ContentDisposition.filename(value) == "broken%"
+    end
+
+    test "falls back to the raw value instead of raising on an invalid hex escape in filename*=" do
+      value = "attachment; filename*=UTF-8''broken%zz.jpg"
+      assert ContentDisposition.filename(value) == "broken%zz.jpg"
+    end
+  end
+end
diff --git a/test/http_client/hackney_test.exs b/test/http_client/hackney_test.exs
index a362bd3..016d758 100644
--- a/test/http_client/hackney_test.exs
+++ b/test/http_client/hackney_test.exs
@@ -1,48 +1,362 @@
 defmodule WaffleTest.HTTPClient.Hackney do
   use ExUnit.Case, async: false
   import Mock
+  import WaffleTest.Support.HackneyMock, only: [mock_hackney_messages: 1]
 
   alias Waffle.HTTPClient.Hackney
 
-  describe "get/3" do
+  describe "get/3 successful responses" do
     test "returns {:ok, body} on 200 with no content-disposition header" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "file content", :done])
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 200, [], :client_ref} end,
-        body: fn :client_ref, :infinity -> {:ok, "file content"} end do
+        get: fn _url, _headers, "", opts ->
+          assert {:async, :once} in opts
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
         result = Hackney.get("http://example.com/file.jpg", [], [])
         assert result == {:ok, "file content"}
       end
     end
 
-    test "returns {:ok, body, filename} when content-disposition header is present" do
+    test "returns {:ok, body, filename} when content-disposition has a quoted filename" do
       response_headers = [{"content-disposition", ~s(attachment; filename="photo.jpg")}]
 
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, response_headers}, "file content", :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file", [], [])
+        assert result == {:ok, "file content", "photo.jpg"}
+      end
+    end
+
+    test "returns {:ok, body, filename} when content-disposition has an unquoted filename" do
+      response_headers = [{"content-disposition", "attachment; filename=photo.jpg"}]
+
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, response_headers}, "file content", :done])
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 200, response_headers, :client_ref} end,
-        body: fn :client_ref, :infinity -> {:ok, "file content"} end do
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
         result = Hackney.get("http://example.com/file", [], [])
         assert result == {:ok, "file content", "photo.jpg"}
       end
     end
 
-    test "returns {:error, :service_unavailable} on 503" do
+    test "returns {:ok, body, filename} decoded from RFC 5987 filename*=" do
+      response_headers = [{"content-disposition", "attachment; filename*=UTF-8''my%20photo.jpg"}]
+
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, response_headers}, "file content", :done])
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 503, [], :client_ref} end,
-        close: fn :client_ref -> :ok end do
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file", [], [])
+        assert result == {:ok, "file content", "my photo.jpg"}
+      end
+    end
+
+    test "prefers filename*= over filename= when both are present" do
+      response_headers = [
+        {"content-disposition",
+         ~s(attachment; filename="fallback.jpg"; filename*=UTF-8''preferred.jpg)}
+      ]
+
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, response_headers}, "file content", :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file", [], [])
+        assert result == {:ok, "file content", "preferred.jpg"}
+      end
+    end
+
+    test "returns {:ok, body} when content-disposition has no filename parameter" do
+      response_headers = [{"content-disposition", "inline"}]
+
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, response_headers}, "file content", :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file", [], [])
+        assert result == {:ok, "file content"}
+      end
+    end
+
+    test "concatenates multiple body chunks in order" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "ab", "cd", "ef", :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file.jpg", [], [])
+        assert result == {:ok, "abcdef"}
+      end
+    end
+  end
+
+  describe "get/3 non-200 responses" do
+    test "returns {:error, :service_unavailable} on 503 and closes the connection" do
+      {ref, send_auto, _send_next} = mock_hackney_messages([{:status, 503, "Service Unavailable"}])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        close: fn ^ref -> :ok end do
         result = Hackney.get("http://example.com/file.jpg", [], [])
         assert result == {:error, :service_unavailable}
+        assert_called(:hackney.close(:_))
       end
     end
 
-    test "returns {:error, {:http_error, status}} with the actual HTTP status on non-200/503" do
+    test "returns {:error, {:http_error, status}} with the actual HTTP status on non-200/503 and closes the connection" do
+      {ref, send_auto, _send_next} = mock_hackney_messages([{:status, 404, "Not Found"}])
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 404, [], :client_ref} end,
-        close: fn :client_ref -> :ok end do
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        close: fn ^ref -> :ok end do
         result = Hackney.get("http://example.com/file.jpg", [], [])
         assert result == {:error, {:http_error, 404}}
+        assert_called(:hackney.close(:_))
+      end
+    end
+  end
+
+  describe "get/3 redirects" do
+    test "follows a 301 redirect and returns the body from the final location" do
+      {ref1, send_auto1, _send_next1} =
+        mock_hackney_messages([{:redirect, "http://example.com/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", _headers, "", _opts ->
+            send_auto1.()
+            {:ok, ref1}
+
+          "http://example.com/final.jpg", _headers, "", _opts ->
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", [], [])
+        assert result == {:ok, "final content"}
+      end
+    end
+
+    test "follows a 303 see_other redirect (e.g. after a POST) the same way as a redirect" do
+      {ref1, send_auto1, _send_next1} =
+        mock_hackney_messages([{:see_other, "http://example.com/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", _headers, "", _opts ->
+            send_auto1.()
+            {:ok, ref1}
+
+          "http://example.com/final.jpg", _headers, "", _opts ->
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", [], [])
+        assert result == {:ok, "final content"}
       end
     end
 
+    test "resolves a relative Location against the original URL" do
+      {ref1, send_auto1, _send_next1} = mock_hackney_messages([{:redirect, "/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", _headers, "", _opts ->
+            send_auto1.()
+            {:ok, ref1}
+
+          "http://example.com/final.jpg", _headers, "", _opts ->
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", [], [])
+        assert result == {:ok, "final content"}
+      end
+    end
+
+    test "strips Authorization and Cookie headers when redirecting to a different origin" do
+      {ref1, send_auto1, _send_next1} =
+        mock_hackney_messages([{:redirect, "https://other.example.com/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      original_headers = [
+        {"authorization", "Bearer secret"},
+        {"cookie", "session=abc"},
+        {"accept", "*/*"}
+      ]
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", headers, "", _opts ->
+            assert {"authorization", "Bearer secret"} in headers
+            send_auto1.()
+            {:ok, ref1}
+
+          "https://other.example.com/final.jpg", headers, "", _opts ->
+            refute Enum.any?(headers, fn {k, _v} -> String.downcase(k) == "authorization" end)
+            refute Enum.any?(headers, fn {k, _v} -> String.downcase(k) == "cookie" end)
+            assert {"accept", "*/*"} in headers
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", original_headers, [])
+        assert result == {:ok, "final content"}
+      end
+    end
+
+    test "keeps headers intact when redirecting within the same origin" do
+      {ref1, send_auto1, _send_next1} =
+        mock_hackney_messages([{:redirect, "http://example.com/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      original_headers = [{"authorization", "Bearer secret"}]
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", _headers, "", _opts ->
+            send_auto1.()
+            {:ok, ref1}
+
+          "http://example.com/final.jpg", headers, "", _opts ->
+            assert {"authorization", "Bearer secret"} in headers
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", original_headers, [])
+        assert result == {:ok, "final content"}
+      end
+    end
+
+    test "gives up after max_redirect hops and returns {:error, {:too_many_redirects, count}}" do
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          ref = make_ref()
+          send(self(), {:hackney_response, ref, {:redirect, "http://example.com/loop", []}})
+          {:ok, ref}
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/loop", [], max_redirect: 2)
+        assert result == {:error, {:too_many_redirects, 2}}
+      end
+    end
+
+    test "does not treat a 301 as a redirect when follow_redirect is false" do
+      {ref, send_auto, _send_next} = mock_hackney_messages([{:status, 301, "Moved Permanently"}])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", opts ->
+          assert Keyword.get(opts, :follow_redirect) == false
+          send_auto.()
+          {:ok, ref}
+        end,
+        close: fn ^ref -> :ok end do
+        result = Hackney.get("http://example.com/file.jpg", [], follow_redirect: false)
+        assert result == {:error, {:http_error, 301}}
+      end
+    end
+  end
+
+  describe "get/3 connection-level errors" do
     test "returns {:error, :timeout} on hackney timeout map" do
       with_mock :hackney,
         get: fn _url, _headers, "", _opts -> {:error, %{reason: :timeout}} end do
@@ -51,6 +365,22 @@ defmodule WaffleTest.HTTPClient.Hackney do
       end
     end
 
+    test "returns {:error, :timeout} on the bare :connect_timeout atom (hackney 4.5.2 shape)" do
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts -> {:error, :connect_timeout} end do
+        result = Hackney.get("http://example.com/file.jpg", [], [])
+        assert result == {:error, :timeout}
+      end
+    end
+
+    test "returns {:error, :timeout} on the bare :checkout_timeout atom (hackney 4.5.2 shape)" do
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts -> {:error, :checkout_timeout} end do
+        result = Hackney.get("http://example.com/file.jpg", [], [])
+        assert result == {:error, :timeout}
+      end
+    end
+
     test "returns {:error, :recv_timeout} on :timeout atom" do
       with_mock :hackney,
         get: fn _url, _headers, "", _opts -> {:error, :timeout} end do
@@ -67,50 +397,245 @@ defmodule WaffleTest.HTTPClient.Hackney do
       end
     end
 
-    test "returns {:error, :timeout} when body read times out on connect-timeout shape" do
+    test "returns {:error, :timeout} when an error arrives instead of headers (defensive/synthetic scenario)" do
+      ref = make_ref()
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 200, [], :client_ref} end,
-        body: fn :client_ref, :infinity -> {:error, %{reason: :timeout}} end,
-        close: fn :client_ref -> :ok end do
+        get: fn _url, _headers, "", _opts ->
+          send(self(), {:hackney_response, ref, {:status, 200, "OK"}})
+          send(self(), {:hackney_response, ref, {:error, %{reason: :timeout}}})
+          {:ok, ref}
+        end,
+        close: fn ^ref -> :ok end do
         result = Hackney.get("http://example.com/file.jpg", [], [])
         assert result == {:error, :timeout}
+        assert_called(:hackney.close(:_))
       end
     end
 
-    test "returns {:error, :recv_timeout} when body read times out on :timeout atom shape" do
+    test "returns {:error, {:http_error, reason}} when an error arrives mid-body-stream" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, {:error, :closed}])
+
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 200, [], :client_ref} end,
-        body: fn :client_ref, :infinity -> {:error, :timeout} end,
-        close: fn :client_ref -> :ok end do
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end,
+        close: fn ^ref -> :ok end do
         result = Hackney.get("http://example.com/file.jpg", [], [])
-        assert result == {:error, :recv_timeout}
+        assert result == {:error, {:http_error, :closed}}
+        assert_called(:hackney.close(:_))
       end
     end
+  end
 
+  describe "get/3 option passthrough" do
     test "passes recv_timeout and connect_timeout to hackney" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "body", :done])
+
       with_mock :hackney,
         get: fn _url, _headers, "", opts ->
           assert Keyword.get(opts, :recv_timeout) == 3_000
           assert Keyword.get(opts, :connect_timeout) == 8_000
-          {:ok, 200, [], :client_ref}
+          send_auto.()
+          {:ok, ref}
         end,
-        body: fn :client_ref, :infinity -> {:ok, "body"} end do
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
         Hackney.get("http://example.com/file.jpg", [],
           recv_timeout: 3_000,
           connect_timeout: 8_000
         )
       end
     end
+  end
+
+  describe "get/3 max_body_length enforcement" do
+    test "returns {:ok, body} when body size is within max_body_length" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "hello", :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file.jpg", [], max_body_length: 10)
+        assert result == {:ok, "hello"}
+      end
+    end
+
+    test "returns {:ok, body} when body size is exactly at max_body_length" do
+      body = String.duplicate("a", 10)
+      {ref, send_auto, send_next} = mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, body, :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end do
+        result = Hackney.get("http://example.com/file.jpg", [], max_body_length: 10)
+        assert result == {:ok, body}
+      end
+    end
+
+    test "aborts the connection and returns {:error, {:http_error, :body_too_large}} when the body exceeds max_body_length" do
+      big_chunk = String.duplicate("a", 20)
+      {ref, send_auto, send_next} = mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, big_chunk, :done])
 
-    test "passes max_body_length to hackney body read" do
       with_mock :hackney,
-        get: fn _url, _headers, "", _opts -> {:ok, 200, [], :client_ref} end,
-        body: fn :client_ref, max ->
-          assert max == 1024
-          {:ok, "body"}
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end,
+        close: fn ^ref -> :ok end do
+        result = Hackney.get("http://example.com/file.jpg", [], max_body_length: 10)
+        assert result == {:error, {:http_error, :body_too_large}}
+        assert_called(:hackney.close(:_))
+      end
+    end
+
+    test "aborts as soon as the cumulative size across multiple chunks exceeds max_body_length, without requesting further chunks" do
+      {ref, send_auto, send_next} =
+        mock_hackney_messages([
+          {:status, 200, "OK"},
+          {:headers, []},
+          "123456",
+          "789012",
+          "should-never-be-sent",
+          :done
+        ])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
+        end,
+        close: fn ^ref -> :ok end do
+        result = Hackney.get("http://example.com/file.jpg", [], max_body_length: 10)
+        assert result == {:error, {:http_error, :body_too_large}}
+        assert_called(:hackney.close(:_))
+      end
+    end
+
+    test "does not enforce a limit when max_body_length is :infinity (default)" do
+      body = String.duplicate("a", 1_000_000)
+      {ref, send_auto, send_next} = mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, body, :done])
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send_auto.()
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          send_next.()
+          :ok
         end do
-        Hackney.get("http://example.com/file.jpg", [], max_body_length: 1024)
+        result = Hackney.get("http://example.com/file.jpg", [], [])
+        assert result == {:ok, body}
+      end
+    end
+  end
+
+  describe "get/3 mailbox hygiene" do
+    test "flushes stray messages left in the mailbox after a 503" do
+      ref = make_ref()
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send(self(), {:hackney_response, ref, {:status, 503, "Service Unavailable"}})
+          send(self(), {:hackney_response, ref, {:headers, []}})
+          {:ok, ref}
+        end,
+        close: fn ^ref -> :ok end do
+        result = Hackney.get("http://example.com/file.jpg", [], [])
+        assert result == {:error, :service_unavailable}
+        refute_receive {:hackney_response, ^ref, _}, 50
       end
     end
+
+    test "flushes a stray trailing message left in the mailbox after a body-too-large abort" do
+      ref = make_ref()
+
+      with_mock :hackney,
+        get: fn _url, _headers, "", _opts ->
+          send(self(), {:hackney_response, ref, {:status, 200, "OK"}})
+          send(self(), {:hackney_response, ref, {:headers, []}})
+          {:ok, ref}
+        end,
+        stream_next: fn ^ref ->
+          # Simulate an oversized chunk arriving together with a subsequent
+          # message that we never get a chance to explicitly ask for.
+          send(self(), {:hackney_response, ref, String.duplicate("a", 20)})
+          send(self(), {:hackney_response, ref, :done})
+          :ok
+        end,
+        close: fn ^ref -> :ok end do
+        result = Hackney.get("http://example.com/file.jpg", [], max_body_length: 10)
+        assert result == {:error, {:http_error, :body_too_large}}
+        refute_receive {:hackney_response, ^ref, _}, 50
+      end
+    end
+
+    test "flushes the redirect message's connection before following it" do
+      {ref1, send_auto1, _send_next1} =
+        mock_hackney_messages([{:redirect, "http://example.com/final.jpg", []}])
+
+      {ref2, send_auto2, send_next2} =
+        mock_hackney_messages([{:status, 200, "OK"}, {:headers, []}, "final content", :done])
+
+      with_mock :hackney,
+        get: fn
+          "http://example.com/original.jpg", _headers, "", _opts ->
+            send_auto1.()
+            {:ok, ref1}
+
+          "http://example.com/final.jpg", _headers, "", _opts ->
+            send_auto2.()
+            {:ok, ref2}
+        end,
+        stream_next: fn ^ref2 ->
+          send_next2.()
+          :ok
+        end,
+        close: fn _ref -> :ok end do
+        result = Hackney.get("http://example.com/original.jpg", [], [])
+        assert result == {:ok, "final content"}
+        refute_receive {:hackney_response, ^ref1, _}, 50
+      end
+    end
+  end
+
+  describe "get/3 real network request (no mocks)" do
+    @describetag :external
+
+    test "fetches https://www.google.com/favicon.ico" do
+      {:ok, body} = Hackney.get("https://www.google.com/favicon.ico", [], [])
+      assert byte_size(body) > 0
+    end
   end
 end
diff --git a/test/support/hackney_mock.ex b/test/support/hackney_mock.ex
new file mode 100644
index 0000000..cad323b
--- /dev/null
+++ b/test/support/hackney_mock.ex
@@ -0,0 +1,54 @@
+defmodule WaffleTest.Support.HackneyMock do
+  @moduledoc """
+  Shared helper for simulating hackney 4.5.2's `{async, :once}` message
+  sequence in tests, used by both `WaffleTest.HTTPClient.Hackney` and
+  `WaffleTest.Actions.Store`.
+
+  In `{async, :once}` mode, hackney pushes `{:status, ...}` and
+  `{:headers, ...}` (or a single `{:redirect, ...}`/`{:see_other, ...}`)
+  unconditionally as soon as the response is parsed -- no `stream_next/1`
+  call is needed to receive them. Only body chunks (and the trailing
+  `:done`/`{:error, _}`) require an explicit `stream_next/1` call per item.
+
+  `mock_hackney_messages/1` takes the full sequence of messages a real
+  connection would send and splits it into:
+
+    * `send_auto` - simulates the messages hackney pushes automatically;
+      invoke from the `:hackney.get/4` mock.
+    * `send_next` - simulates what a `stream_next/1` call releases, popping
+      one queued message per call; invoke from the `:hackney.stream_next/1`
+      mock.
+  """
+
+  @doc """
+  Returns `{ref, send_auto, send_next}` for the given message sequence.
+  """
+  def mock_hackney_messages(messages) do
+    ref = make_ref()
+    test_pid = self()
+    {auto, rest} = Enum.split_while(messages, &auto_message?/1)
+    {:ok, agent} = Agent.start_link(fn -> rest end)
+
+    send_auto = fn ->
+      Enum.each(auto, &send(test_pid, {:hackney_response, ref, &1}))
+    end
+
+    send_next = fn ->
+      case Agent.get_and_update(agent, fn
+             [next | rest] -> {next, rest}
+             [] -> {nil, []}
+           end) do
+        nil -> :ok
+        msg -> send(test_pid, {:hackney_response, ref, msg})
+      end
+    end
+
+    {ref, send_auto, send_next}
+  end
+
+  defp auto_message?({:status, _, _}), do: true
+  defp auto_message?({:headers, _}), do: true
+  defp auto_message?({:redirect, _, _}), do: true
+  defp auto_message?({:see_other, _, _}), do: true
+  defp auto_message?(_), do: false
+end