From 4a91e43ec49c3db63f6c6b1ece383e246e95254b Mon Sep 17 00:00:00 2001 From: Ruslan Levond Date: Fri, 10 Jul 2026 17:02:14 +0100 Subject: [PATCH 1/2] Fix EventStream async streaming under hackney 4 With hackney 4.x the streaming path hung indefinitely: the stream opened but no chunks were ever delivered. Two incompatibilities, plus two hardening fixes: - Force HTTP/1.1 (protocols: [:http1]). hackney 4 negotiates HTTP/2 via ALPN by default and Bedrock accepts h2, but hackney's h2 path never delivers async body messages, so the first receive (status) blocked forever. HTTP/1.1 is the documented protocol for InvokeModelWithResponseStream and what boto3 uses; it also guarantees the Transfer-Encoding: chunked header this module verifies. - Accept pid async refs. hackney 4 correlates async messages by the connection pid (async_ref :: pid()), not an Erlang reference, so the `ref when is_reference(ref)` clause raised FunctionClauseError on the first body read. Match on reference or pid to support both majors. - Honor caller :http_opts (recv_timeout/connect_timeout/pool) via hackney_options/1, streaming defaults winning on conflict. Previously the stream always used hackney's built-in recv_timeout (5s) and dropped slow responses regardless of the caller's configured timeout. - Raise on {:error, reason} async messages. hackney 4 emits e.g. {:error, :closed} on mid-stream disconnects; previously any shape other than {:closed, :timeout} either blocked the receive forever or fell into the body-data clause and crashed decode with a badarg. Guard the data clause with is_binary/1 and raise ExAws.Error with the real reason instead. Verified live against eu.anthropic.claude-sonnet-4-5-20250929-v1:0 (eu-west-1) with hackney 4.5.2: full stream decoded, zero bad chunks. --- lib/ex_aws/bedrock/event_stream.ex | 41 ++++++++++++++++++++--- test/ex_aws/bedrock/event_stream_test.exs | 29 ++++++++++++++++ 2 files changed, 66 insertions(+), 4 deletions(-) diff --git a/lib/ex_aws/bedrock/event_stream.ex b/lib/ex_aws/bedrock/event_stream.ex index 1816499..8fa00f3 100644 --- a/lib/ex_aws/bedrock/event_stream.ex +++ b/lib/ex_aws/bedrock/event_stream.ex @@ -33,7 +33,11 @@ defmodule ExAws.Bedrock.EventStream do {"user-agent", @user_agent}, {"x-amzn-bedrock-accept", "*/*"} ] - @hackney_options [{:async, :once}] + # :protocols forces HTTP/1.1: hackney >= 4 negotiates HTTP/2 via ALPN by + # default, but its h2 path does not deliver async body messages, so the + # event stream would hang waiting for chunks that never arrive. HTTP/1.1 + # also guarantees the chunked transfer-encoding this module verifies. + @hackney_options [{:async, :once}, {:protocols, [:http1]}] @doc """ Stream of chunks from the response stream. @@ -57,8 +61,10 @@ defmodule ExAws.Bedrock.EventStream do encoded_data ) + hackney_opts = hackney_options(config) + request_fun = fn [] -> - {:ok, ref} = :hackney.post(url, full_headers, encoded_data, @hackney_options) + {:ok, ref} = :hackney.post(url, full_headers, encoded_data, hackney_opts) receive do {:hackney_response, ^ref, {:status, 200, _reason}} -> @@ -69,6 +75,9 @@ defmodule ExAws.Bedrock.EventStream do {:hackney_response, ^ref, {:error, {:closed, :timeout}}} -> :closed + + {:hackney_response, ^ref, {:error, reason}} -> + raise ExAws.Error, "Bedrock stream request failed: #{inspect(reason)}" end end @@ -82,7 +91,9 @@ defmodule ExAws.Bedrock.EventStream do {:error, status, reason} -> raise ExAws.Error, "#{to_string(status)}: #{to_string(reason)}" - ref when is_reference(ref) -> + # hackney < 4 identifies async responses by reference, + # hackney >= 4 by the connection pid. + ref when is_reference(ref) or is_pid(ref) -> :ok = :hackney.stream_next(ref) receive do @@ -94,7 +105,10 @@ defmodule ExAws.Bedrock.EventStream do {:hackney_response, ^ref, :done} -> {:halt, []} - {:hackney_response, ^ref, data} -> + {:hackney_response, ^ref, {:error, reason}} -> + raise ExAws.Error, "Bedrock stream failed mid-stream: #{inspect(reason)}" + + {:hackney_response, ^ref, data} when is_binary(data) -> {[data], ref} end end, @@ -104,6 +118,25 @@ defmodule ExAws.Bedrock.EventStream do Stream.flat_map(stream, &decode_chunk/1) end + @doc """ + Builds the hackney options for the streaming request. + + Merges caller-provided options from the ExAws config `:http_opts` (e.g. + `recv_timeout`, `connect_timeout`, `pool`) on top of the async-streaming + defaults. Without this, the stream would always use hackney's built-in + `recv_timeout` (5s) and drop slow responses regardless of the timeout the + caller configured. + + The streaming defaults win on conflicting keys, so the async-streaming mode + (`async: :once`) and the forced HTTP/1.1 protocol can't be accidentally + disabled by caller options. + """ + def hackney_options(config) do + config + |> Map.get(:http_opts, []) + |> Keyword.merge(@hackney_options) + end + defp verify_event_stream!(headers) do verify_header!(headers, "Content-Type", @content_type) end diff --git a/test/ex_aws/bedrock/event_stream_test.exs b/test/ex_aws/bedrock/event_stream_test.exs index 2977bc0..a2c27a4 100644 --- a/test/ex_aws/bedrock/event_stream_test.exs +++ b/test/ex_aws/bedrock/event_stream_test.exs @@ -34,6 +34,35 @@ defmodule ExAws.Bedrock.EventStreamTest do end end + describe "hackney_options/1" do + test "defaults to async streaming over HTTP/1.1 when no http_opts are given" do + assert EventStream.hackney_options(%{}) == [async: :once, protocols: [:http1]] + end + + test "honors caller recv_timeout / connect_timeout / pool from :http_opts" do + opts = + EventStream.hackney_options(%{ + http_opts: [recv_timeout: 600_000, connect_timeout: 10_000, pool: :ex_aws] + }) + + assert Keyword.get(opts, :async) == :once + assert Keyword.get(opts, :recv_timeout) == 600_000 + assert Keyword.get(opts, :connect_timeout) == 10_000 + assert Keyword.get(opts, :pool) == :ex_aws + end + + test "streaming defaults win so async: :once and HTTP/1.1 cannot be disabled" do + opts = + EventStream.hackney_options(%{ + http_opts: [async: false, protocols: [:http2], recv_timeout: 1_000] + }) + + assert Keyword.get(opts, :async) == :once + assert Keyword.get(opts, :protocols) == [:http1] + assert Keyword.get(opts, :recv_timeout) == 1_000 + end + end + setup_all do # Claude 3.5 Sonnet Multi-Chunk response multipart_chunk = From 6841752f7a2e2e21b175d0ad9e4099fed4ad3c5b Mon Sep 17 00:00:00 2001 From: Ruslan Levond Date: Tue, 14 Jul 2026 17:40:45 +0100 Subject: [PATCH 2/2] Updated EventStream with a defensive catch-all clause in streaming --- lib/ex_aws/bedrock/event_stream.ex | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/lib/ex_aws/bedrock/event_stream.ex b/lib/ex_aws/bedrock/event_stream.ex index 8fa00f3..91cd87f 100644 --- a/lib/ex_aws/bedrock/event_stream.ex +++ b/lib/ex_aws/bedrock/event_stream.ex @@ -110,6 +110,10 @@ defmodule ExAws.Bedrock.EventStream do {:hackney_response, ^ref, data} when is_binary(data) -> {[data], ref} + + {:hackney_response, ^ref, other} -> + raise ExAws.Error, + "Bedrock stream received unexpected message: #{inspect(other)}" end end, &Function.identity/1