From f27e209f83825b93409cc0446e378676f33938b4 Mon Sep 17 00:00:00 2001
From: Benoit Chesneau <bchesneau@gmail.com>
Date: Tue, 9 Jun 2026 14:24:56 +0200
Subject: [PATCH 1/3] port to OTP 28/29 and hackney 4.2.2

- replace deprecated standalone catch with try/catch (OTP 29)
- migrate to hackney 4.2.2: eager response bodies, drop skip_body/1
- rework pull-based streaming downloads (multipart open_doc,
  fetch_attachment/stream_attachment, stream_doc) onto hackney async mode
- bump meck to 1.2.0 for OTP 29, update test mocks to eager-body shapes
- add OTP 29 to CI matrix, bump version to 2.1.0
---
 .github/workflows/test.yml |   4 +-
 CHANGELOG.md               |  26 +++++
 rebar.config               |   4 +-
 src/couchbeam.app.src      |   2 +-
 src/couchbeam.erl          | 216 +++++++++++++++++++------------------
 src/couchbeam_changes.erl  |  37 +++----
 src/couchbeam_httpc.erl    | 176 +++++++++++++++++++++---------
 src/couchbeam_util.erl     |   4 +-
 src/couchbeam_uuids.erl    |   6 +-
 src/couchbeam_view.erl     |   9 +-
 test/couchbeam_mocks.erl   |  65 ++++-------
 11 files changed, 309 insertions(+), 240 deletions(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index cb3919f3..490332ee 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -14,7 +14,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        otp: ['27.3', '28.0']
+        otp: ['27.3', '28.0', '29.0']
         rebar3: ['3.25.0']
 
     steps:
@@ -55,7 +55,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        otp: ['27.3', '28.0']
+        otp: ['27.3', '28.0', '29.0']
         rebar3: ['3.25.0']
 
     steps:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ca7a0cf7..6b471f92 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,32 @@
 
 All notable changes to this project will be documented in this file.
 
+## [2.1.0] - 2026-06-09
+
+### Changed
+
+- Updated to hackney 4.2.2. Response bodies are now read eagerly: a normal
+  request returns `{ok, Status, Headers, Body}` with the body as a binary.
+- Reworked pull-based response streaming (multipart `open_doc`,
+  `fetch_attachment`/`stream_attachment` with `stream`, `stream_doc`) onto
+  hackney's async mode. The public streaming API is unchanged.
+- Replaced deprecated standalone `catch` expressions with `try ... catch`
+  for OTP 29.
+
+### Compatibility
+
+- Supports OTP 27, 28 and 29.
+- CI now also runs on OTP 29.
+
+### Dependencies
+
+- hackney: 4.2.2 (from 2.0.1)
+- meck (test): 1.2.0 (from 0.9.2)
+
+### Removed
+
+- `hackney:skip_body/1` usage (removed in hackney 4.x); bodies are read directly.
+
 ## [2.0.0] - 2026-01-21
 
 ### Breaking Changes
diff --git a/rebar.config b/rebar.config
index b49aa4d1..a694c5ca 100644
--- a/rebar.config
+++ b/rebar.config
@@ -17,7 +17,7 @@
 ]}.
 
 {deps, [
-    {hackney, "2.0.1"},
+    {hackney, "4.2.2"},
     %% oauth is optional - only needed if using OAuth authentication
     {oauth, "2.1.0"}
 ]}.
@@ -58,6 +58,6 @@
 {profiles, [{test, [
                     {cover_enabled, true},
                     {eunit_opts, [verbose]},
-                    {deps, [{oauth, "2.1.0"}, {meck, "0.9.2"}]}
+                    {deps, [{oauth, "2.1.0"}, {meck, "1.2.0"}]}
                    ]}
            ]}.
diff --git a/src/couchbeam.app.src b/src/couchbeam.app.src
index 48fd8df7..94bb4bd1 100644
--- a/src/couchbeam.app.src
+++ b/src/couchbeam.app.src
@@ -6,7 +6,7 @@
 
 {application, couchbeam,
  [{description, "Erlang CouchDB client"},
-  {vsn, "2.0.0"},
+  {vsn, "2.1.0"},
   {modules, []},
   {registered, [
     couchbeam_sup
diff --git a/src/couchbeam.erl b/src/couchbeam.erl
index b43e1319..567df899 100644
--- a/src/couchbeam.erl
+++ b/src/couchbeam.erl
@@ -154,11 +154,9 @@ server_connection(Host, Port, Prefix, Options) ->
 -spec server_info(server()) -> {ok, map()} | {error, term()}.
 server_info(#server{url=Url, options=Opts}) ->
     case hackney:get(Url, [], <<>>, Opts) of
-        {ok, 200, _, Ref} ->
-            Version = couchbeam_httpc:json_body(Ref),
-            {ok, Version};
-        {ok, Status, Headers, Ref} ->
-            {ok, Body} = hackney:body(Ref),
+        {ok, 200, _, Body} ->
+            {ok, couchbeam_httpc:json_body(Body)};
+        {ok, Status, Headers, Body} ->
             {error, {bad_response, {Status, Headers, Body}}};
 
         Error ->
@@ -272,8 +270,7 @@ view_cleanup(#db{server=Server, name=DbName, options=Opts}) ->
     Headers = [{<<"Content-Type">>, <<"application/json">>}],
     Resp = couchbeam_httpc:db_request(post, Url, Headers, <<>>, Opts, [200, 202]),
     case Resp of
-        {ok, _, _, Ref} ->
-            catch hackney:skip_body(Ref),
+        {ok, _, _, _} ->
             ok;
         Error ->
             Error
@@ -318,8 +315,7 @@ create_db(#server{url=ServerUrl, options=Opts}=Server, DbName0, Options,
     Url = hackney_url:make_url(ServerUrl, DbName, Params),
     Resp = couchbeam_httpc:db_request(put, Url, [], <<>>, Options1, [201]),
     case Resp of
-        {ok, _Status, _Headers, Ref} ->
-            hackney:skip_body(Ref),
+        {ok, _Status, _Headers, _} ->
             {ok, #db{server=Server, name=DbName, options=Options1}};
         {error, precondition_failed} ->
             {error, db_exists};
@@ -362,8 +358,7 @@ open_or_create_db(#server{url=ServerUrl, options=Opts}=Server, DbName0,
     Opts1 = couchbeam_util:propmerge1(Options, Opts),
     Resp = couchbeam_httpc:request(get, Url, [], <<>>, Opts1),
     case couchbeam_httpc:db_resp(Resp, [200]) of
-        {ok, _Status, _Headers, Ref} ->
-            hackney:skip_body(Ref),
+        {ok, _Status, _Headers, _} ->
             open_db(Server, DbName, Options);
         {error, not_found} ->
             create_db(Server, DbName, Options, Params);
@@ -429,39 +424,72 @@ open_doc(#db{server=Server, options=Opts}=Db, DocId, Params) ->
                             A -> {A, proplists:delete(accept, Params)}
                         end,
     %% set the headers with the accepted content-type if needed
-    Headers = case {Accept, proplists:get_value("attachments", Params)} of
-                  {any, true} ->
-                      %% only use the more efficient method when we get the
-                      %% attachments so we don't use much bandwidth.
-                      [{<<"Accept">>, <<"multipart/related">>}];
-                  {Accept, _} when is_binary(Accept) ->
-                      %% accepted content-type has been forced
-                      [{<<"Accept">>, Accept}];
-                  _ ->
-                      []
-              end,
+    {Headers, Multipart} =
+        case {Accept, proplists:get_value("attachments", Params)} of
+            {any, true} ->
+                %% only use the more efficient method when we get the
+                %% attachments so we don't use much bandwidth.
+                {[{<<"Accept">>, <<"multipart/related">>}], true};
+            {Accept, _} when is_binary(Accept) ->
+                %% accepted content-type has been forced
+                {[{<<"Accept">>, Accept}], is_multipart_accept(Accept)};
+            _ ->
+                {[], false}
+        end,
     Url = hackney_url:make_url(couchbeam_httpc:server_url(Server), couchbeam_httpc:doc_url(Db, DocId1),
                                Params1),
-    case couchbeam_httpc:db_request(get, Url, Headers, <<>>, Opts,
-                                    [200, 201]) of
-        {ok, _, RespHeaders, Ref} ->
+    case Multipart of
+        true ->
+            %% a multipart response is possible: stream it so attachments can
+            %% be read incrementally.
+            open_doc_multipart(Url, Headers, Opts);
+        false ->
+            case couchbeam_httpc:db_request(get, Url, Headers, <<>>, Opts,
+                                            [200, 201]) of
+                {ok, _, _, Body} ->
+                    {ok, couchbeam_httpc:json_body(Body)};
+                Error ->
+                    Error
+            end
+    end.
+
+is_multipart_accept(Accept) ->
+    binary:match(Accept, <<"multipart">>) =/= nomatch.
+
+open_doc_multipart(Url, Headers, Opts) ->
+    case couchbeam_httpc:stream_request(get, Url, Headers, <<>>, Opts) of
+        {ok, Status, RespHeaders, Ref} when Status =:= 200 orelse Status =:= 201 ->
             ParsedHeaders = hackney_headers:from_list(RespHeaders),
             case hackney_headers:get_value(<<"content-type">>, ParsedHeaders) of
                 undefined ->
-                    {ok, couchbeam_httpc:json_body(Ref)};
+                    decode_stream_doc(Ref);
                 ContentType ->
                     case hackney_headers:parse_content_type(ContentType) of
-                        {<<"multipart">>, _, Params} ->
-                            %% we get a multipart request, start to parse it.
-                            {_, Boundary} = lists:keyfind(<<"boundary">>, 1, Params),
-                            InitialState =  {Ref, fun() ->
-                                                          couchbeam_httpc:wait_mp_doc(Ref, Boundary, <<>>)
-                                                  end},
+                        {<<"multipart">>, _, CTParams} ->
+                            %% we get a multipart response, start to parse it.
+                            {_, Boundary} = lists:keyfind(<<"boundary">>, 1, CTParams),
+                            InitialState = {Ref, fun() ->
+                                                         couchbeam_httpc:wait_mp_doc(Ref, Boundary, <<>>)
+                                                 end},
                             {ok, {multipart, InitialState}};
                         _ ->
-                            {ok, couchbeam_httpc:json_body(Ref)}
+                            decode_stream_doc(Ref)
                     end
             end;
+        {ok, 404, _, Ref} ->
+            _ = couchbeam_httpc:stream_body_all(Ref),
+            {error, not_found};
+        {ok, Status, RespHeaders, Ref} ->
+            {ok, Body} = couchbeam_httpc:stream_body_all(Ref),
+            {error, {bad_response, {Status, RespHeaders, Body}}};
+        Error ->
+            Error
+    end.
+
+decode_stream_doc(Ref) ->
+    case couchbeam_httpc:stream_body_all(Ref) of
+        {ok, Body} ->
+            {ok, couchbeam_httpc:json_body(Body)};
         Error ->
             Error
     end.
@@ -786,20 +814,33 @@ fetch_attachment(#db{server=Server, options=Opts}=Db, DocId, Name, Options0) ->
                                [couchbeam_httpc:db_url(Db), DocId1,
                                 Name],
                                Options2),
-    case hackney:get(Url, Headers, <<>>, Opts) of
-        {ok, 200, _, Ref} when Stream /= true ->
-            hackney:body(Ref);
-        {ok, 200, _, Ref} ->
-            {ok, Ref};
-        {ok, 404, _, Ref} ->
-            hackney:skip_body(Ref),
-            {error, not_found};
-        {ok, Status, Headers, Ref} ->
-            {ok, Body} = hackney:body(Ref),
-            {error, {bad_response, {Status, Headers, Body}}};
-
-        Error ->
-            Error
+    case Stream of
+        true ->
+            %% stream the attachment: drive hackney's async mode and let the
+            %% caller pull chunks with stream_attachment/1.
+            case couchbeam_httpc:stream_request(get, Url, Headers, <<>>, Opts) of
+                {ok, 200, _, Ref} ->
+                    {ok, Ref};
+                {ok, 404, _, Ref} ->
+                    _ = couchbeam_httpc:stream_body_all(Ref),
+                    {error, not_found};
+                {ok, Status, RespHeaders, Ref} ->
+                    {ok, Body} = couchbeam_httpc:stream_body_all(Ref),
+                    {error, {bad_response, {Status, RespHeaders, Body}}};
+                Error ->
+                    Error
+            end;
+        false ->
+            case hackney:get(Url, Headers, <<>>, Opts) of
+                {ok, 200, _, Body} ->
+                    {ok, Body};
+                {ok, 404, _, _} ->
+                    {error, not_found};
+                {ok, Status, RespHeaders, Body} ->
+                    {error, {bad_response, {Status, RespHeaders, Body}}};
+                Error ->
+                    Error
+            end
     end.
 
 %% @doc fetch an attachment chunk.
@@ -820,7 +861,7 @@ fetch_attachment(#db{server=Server, options=Opts}=Db, DocId, Name, Options0) ->
               | done
               | {error, term()}.
 stream_attachment(Ref) ->
-    hackney:stream_body(Ref).
+    couchbeam_httpc:stream_body(Ref).
 
 %% @doc Start streaming an attachment. Returns a reference that can be
 %% passed to stream_attachment/1 to receive chunks.
@@ -883,8 +924,7 @@ put_attachment(#db{server=Server, options=Opts}=Db, DocId, Name, Body,
 send_attachment(Ref, eof) ->
     case hackney:finish_send_body(Ref) of
         ok ->
-            Resp =  hackney:start_response(Ref),
-            couchbeam_httpc:reply_att(Resp);
+            couchbeam_httpc:reply_att(couchbeam_httpc:read_response(Ref));
         Error ->
             Error
     end;
@@ -971,8 +1011,7 @@ compact(#db{server=Server, options=Opts}=Db) ->
                                []),
     Headers = [{<<"Content-Type">>, <<"application/json">>}],
     case couchbeam_httpc:db_request(post, Url, Headers, <<>>, Opts, [202]) of
-        {ok, _, _, Ref} ->
-            hackney:skip_body(Ref),
+        {ok, _, _, _} ->
             ok;
         Error ->
             Error
@@ -987,8 +1026,7 @@ compact(#db{server=Server, options=Opts}=Db, DesignName) ->
                                                                     DesignName], []),
     Headers = [{<<"Content-Type">>, <<"application/json">>}],
     case couchbeam_httpc:db_request(post, Url, Headers, <<>>, Opts, [202]) of
-        {ok, _, _, Ref} ->
-            hackney:skip_body(Ref),
+        {ok, _, _, _} ->
             ok;
         Error ->
             Error
@@ -1087,10 +1125,8 @@ basic_test() ->
         ServerInfoResponse = #{<<"couchdb">> => <<"Welcome">>,
                                <<"version">> => <<"3.3.0">>,
                                <<"uuid">> => <<"test-uuid">>},
-        Ref = make_ref(),
         meck:expect(hackney, get, fun(_Url, _Headers, _Body, _Opts) ->
-            couchbeam_mocks:set_body(Ref, ServerInfoResponse),
-            {ok, 200, [], Ref}
+            {ok, 200, [], couchbeam_mocks:body(ServerInfoResponse)}
         end),
 
         Server = couchbeam:server_connection(),
@@ -1123,9 +1159,7 @@ db_test() ->
                 case lists:member(DbName, Dbs) of
                     true ->
                         put(mock_dbs, lists:delete(DbName, Dbs)),
-                        Ref = make_ref(),
-                        couchbeam_mocks:set_body(Ref, #{<<"ok">> => true}),
-                        {ok, 200, [], Ref};
+                        {ok, 200, [], couchbeam_mocks:body(#{<<"ok">> => true})};
                     false ->
                         {ok, 404, [], make_ref()}
                 end;
@@ -1135,9 +1169,7 @@ db_test() ->
                 Dbs = get(mock_dbs),
                 case lists:member(DbName, Dbs) of
                     true ->
-                        Ref = make_ref(),
-                        couchbeam_mocks:set_body(Ref, #{<<"db_name">> => DbName}),
-                        {ok, 200, [], Ref};
+                        {ok, 200, [], couchbeam_mocks:body(#{<<"db_name">> => DbName})};
                     false ->
                         {ok, 404, [], make_ref()}
                 end;
@@ -1203,15 +1235,11 @@ db_mock_handle_request(get, Url) ->
     case binary:match(Url, <<"_all_dbs">>) of
         nomatch ->
             %% db_info request
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"db_name">> => <<"testdb">>}),
-            {ok, 200, [], Ref};
+            {ok, 200, [], couchbeam_mocks:body(#{<<"db_name">> => <<"testdb">>})};
         _ ->
             %% all_dbs request
             Dbs = get(mock_dbs),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, Dbs),
-            {ok, 200, [], Ref}
+            {ok, 200, [], couchbeam_mocks:body(Dbs)}
     end;
 db_mock_handle_request(_Method, _Url) ->
     {ok, 200, [], make_ref()}.
@@ -1314,7 +1342,7 @@ basic_doc_test() ->
     after
         erase(mock_docs),
         erase(mock_uuid_counter),
-        catch meck:unload(couchbeam_uuids),
+        try meck:unload(couchbeam_uuids) catch _:_ -> ok end,
         couchbeam_mocks:teardown()
     end.
 
@@ -1351,9 +1379,7 @@ doc_mock_handle_request(put, Url, _Headers, Body) ->
                     UpdatedDoc = maps:merge(maps:remove(<<"_rev">>, maps:remove(<<"_id">>, DocProps)),
                                             #{<<"_id">> => DocId, <<"_rev">> => NewRev}),
                     put(mock_docs, maps:put(DocId, {UpdatedDoc, NewRev}, Docs)),
-                    Ref = make_ref(),
-                    couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev}),
-                    {ok, 201, [], Ref}
+                    {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev})}
             end;
         false ->
             %% New document
@@ -1361,9 +1387,7 @@ doc_mock_handle_request(put, Url, _Headers, Body) ->
             NewDoc = maps:merge(maps:remove(<<"_rev">>, maps:remove(<<"_id">>, DocProps)),
                                 #{<<"_id">> => DocId, <<"_rev">> => NewRev}),
             put(mock_docs, maps:put(DocId, {NewDoc, NewRev}, Docs)),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev}),
-            {ok, 201, [], Ref}
+            {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev})}
     end;
 doc_mock_handle_request(get, Url, _Headers, _Body) ->
     %% open_doc - GET /db/docid
@@ -1373,9 +1397,7 @@ doc_mock_handle_request(get, Url, _Headers, _Body) ->
         undefined ->
             {error, not_found};
         {Doc, _Rev} ->
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, Doc),
-            {ok, 200, [], Ref}
+            {ok, 200, [], couchbeam_mocks:body(Doc)}
     end;
 doc_mock_handle_request(head, Url, _Headers, _Body) ->
     %% lookup_doc_rev or doc_exists - HEAD /db/docid
@@ -1411,9 +1433,7 @@ doc_mock_handle_request(post, Url, _Headers, Body) ->
                         #{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev}
                 end
             end, InputDocs),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, Results),
-            {ok, 201, [], Ref}
+            {ok, 201, [], couchbeam_mocks:body(Results)}
     end;
 doc_mock_handle_request(copy, Url, Headers, _Body) ->
     %% copy_doc - COPY /db/docid with Destination header
@@ -1438,9 +1458,7 @@ doc_mock_handle_request(copy, Url, Headers, _Body) ->
             NewDoc = maps:merge(maps:without([<<"_rev">>, <<"_id">>], SourceDoc),
                                 #{<<"_id">> => DestId, <<"_rev">> => NewRev}),
             put(mock_docs, maps:put(DestId, {NewDoc, NewRev}, get(mock_docs))),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"id">> => DestId, <<"rev">> => NewRev}),
-            {ok, 201, [], Ref}
+            {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"id">> => DestId, <<"rev">> => NewRev})}
     end;
 doc_mock_handle_request(_Method, _Url, _Headers, _Body) ->
     {ok, 200, [], make_ref()}.
@@ -1549,7 +1567,7 @@ copy_doc_test() ->
     after
         erase(mock_docs),
         erase(mock_uuid_counter),
-        catch meck:unload(couchbeam_uuids),
+        try meck:unload(couchbeam_uuids) catch _:_ -> ok end,
         couchbeam_mocks:teardown()
     end.
 
@@ -1577,9 +1595,7 @@ attachments_test() ->
                         undefined ->
                             {ok, 404, [], make_ref()};
                         AttBody ->
-                            Ref = make_ref(),
-                            couchbeam_mocks:set_body(Ref, AttBody),
-                            {ok, 200, [], Ref}
+                            {ok, 200, [], couchbeam_mocks:body(AttBody)}
                     end;
                 _ ->
                     {ok, 404, [], make_ref()}
@@ -1633,9 +1649,7 @@ att_mock_handle_request(put, Url, _Headers, Body) ->
             BodyBin = if is_binary(Body) -> Body; true -> iolist_to_binary(Body) end,
             put(mock_attachments, maps:put({DocId, AttName}, BodyBin, Atts)),
             NewRev = generate_mock_rev(),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev}),
-            {ok, 201, [], Ref};
+            {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev})};
         false ->
             %% Regular doc PUT
             doc_mock_handle_request(put, Url, [], Body)
@@ -1646,9 +1660,7 @@ att_mock_handle_request(delete, Url, _Headers, _Body) ->
             Atts = get(mock_attachments),
             put(mock_attachments, maps:remove({DocId, AttName}, Atts)),
             NewRev = generate_mock_rev(),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"rev">> => NewRev}),
-            {ok, 200, [], Ref};
+            {ok, 200, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"rev">> => NewRev})};
         false ->
             doc_mock_handle_request(delete, Url, [], <<>>)
     end;
@@ -1722,7 +1734,7 @@ replicate_test() ->
     after
         erase(mock_docs),
         erase(mock_uuid_counter),
-        catch meck:unload(couchbeam_uuids),
+        try meck:unload(couchbeam_uuids) catch _:_ -> ok end,
         couchbeam_mocks:teardown()
     end.
 
@@ -1736,9 +1748,7 @@ replicate_mock_handle(put, Url, Body) ->
             %% Save replication doc to _replicator
             DocId = generate_mock_uuid(),
             NewRev = generate_mock_rev(),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev}),
-            {ok, 201, [], Ref}
+            {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"id">> => DocId, <<"rev">> => NewRev})}
     end;
 replicate_mock_handle(post, Url, Body) ->
     UrlBin = iolist_to_binary(Url),
@@ -1749,9 +1759,7 @@ replicate_mock_handle(post, Url, Body) ->
                     {ok, 200, [], make_ref()};
                 _ ->
                     %% ensure_full_commit
-                    Ref = make_ref(),
-                    couchbeam_mocks:set_body(Ref, #{<<"ok">> => true, <<"instance_start_time">> => <<"0">>}),
-                    {ok, 201, [], Ref}
+                    {ok, 201, [], couchbeam_mocks:body(#{<<"ok">> => true, <<"instance_start_time">> => <<"0">>})}
             end;
         _ ->
             %% _revs_diff - return all revisions as missing
@@ -1759,9 +1767,7 @@ replicate_mock_handle(post, Url, Body) ->
             Result = maps:fold(fun(DocId, Revs, Acc) ->
                 maps:put(DocId, #{<<"missing">> => Revs}, Acc)
             end, #{}, IdRevs),
-            Ref = make_ref(),
-            couchbeam_mocks:set_body(Ref, Result),
-            {ok, 200, [], Ref}
+            {ok, 200, [], couchbeam_mocks:body(Result)}
     end;
 replicate_mock_handle(Method, Url, Body) ->
     doc_mock_handle_request(Method, Url, [], Body).
diff --git a/src/couchbeam_changes.erl b/src/couchbeam_changes.erl
index e3865aec..7fa08ec5 100644
--- a/src/couchbeam_changes.erl
+++ b/src/couchbeam_changes.erl
@@ -339,9 +339,8 @@ decode_data(Data, #state{feed_type = longpoll, parser = Parser,
     State1 = lists:foldl(fun send_change/2, State, Changes),
     case Parser1 of
         #st{phase = done} ->
-            %% All results parsed
-            catch hackney:stop_async(ClientRef),
-            catch hackney:skip_body(ClientRef),
+            %% All results parsed - close this connection before reconnecting
+            try hackney:close(ClientRef) catch _:_ -> ok end,
             maybe_reconnect(State1#state{parser = Parser1});
         _ ->
             maybe_continue(State1#state{parser = Parser1})
@@ -356,14 +355,16 @@ parse_lines(Buffer, State) ->
     case binary:split(Buffer, <<"\n">>) of
         [Line, Rest] when byte_size(Line) > 0 ->
             %% Got a complete line - parse it
-            case catch couchbeam_ejson:decode(Line) of
-                {'EXIT', _} ->
-                    %% Invalid JSON, skip this line
-                    parse_lines(Rest, State);
+            try couchbeam_ejson:decode(Line) of
                 Change when is_map(Change) ->
                     State1 = send_change(Change, State),
                     parse_lines(Rest, State1);
                 _ ->
+                    %% Not a JSON object, skip this line
+                    parse_lines(Rest, State)
+            catch
+                _:_ ->
+                    %% Invalid JSON, skip this line
                     parse_lines(Rest, State)
             end;
         [<<>>, Rest] ->
@@ -569,19 +570,15 @@ follow_once_test() ->
                     nomatch ->
                         {error, not_found};
                     _ ->
-                        Ref = make_ref(),
-                        meck:expect(hackney, body, fun(_) ->
-                            Changes = [
-                                #{<<"seq">> => 1, <<"id">> => <<"doc1">>, <<"changes">> => [#{<<"rev">> => <<"1-abc">>}]},
-                                #{<<"seq">> => 2, <<"id">> => <<"doc2">>, <<"changes">> => [#{<<"rev">> => <<"1-def">>}]}
-                            ],
-                            Body = couchbeam_ejson:encode(#{
-                                <<"results">> => Changes,
-                                <<"last_seq">> => 2
-                            }),
-                            {ok, Body}
-                        end),
-                        {ok, 200, [], Ref}
+                        Changes = [
+                            #{<<"seq">> => 1, <<"id">> => <<"doc1">>, <<"changes">> => [#{<<"rev">> => <<"1-abc">>}]},
+                            #{<<"seq">> => 2, <<"id">> => <<"doc2">>, <<"changes">> => [#{<<"rev">> => <<"1-def">>}]}
+                        ],
+                        Body = couchbeam_ejson:encode(#{
+                            <<"results">> => Changes,
+                            <<"last_seq">> => 2
+                        }),
+                        {ok, 200, [], Body}
                 end
             end),
 
diff --git a/src/couchbeam_httpc.erl b/src/couchbeam_httpc.erl
index 89e6b41c..a391e25e 100644
--- a/src/couchbeam_httpc.erl
+++ b/src/couchbeam_httpc.erl
@@ -6,6 +6,10 @@
 -module(couchbeam_httpc).
 
 -export([request/5,
+         stream_request/5,
+         stream_body/1,
+         stream_body_all/1,
+         read_response/1,
          db_request/5, db_request/6,
          json_body/1,
          db_resp/2,
@@ -19,13 +23,14 @@
 -include("couchbeam.hrl").
 
 %% @doc Make an HTTP request via hackney.
-%% Returns vary based on options:
-%% - Normal request: {ok, Status, Headers, Pid}
+%% With hackney 4.x the response body is read eagerly. Returns vary based
+%% on options:
+%% - Normal request: {ok, Status, Headers, Body::binary()}
 %% - HEAD request: {ok, Status, Headers}
-%% - Streaming body (body=stream): {ok, Pid}
-%% - Async request: {ok, Pid}
+%% - Streaming body upload (body=stream): {ok, Pid}
+%% - Async request: {ok, Ref}
 -spec request(atom(), binary(), list(), term(), list()) ->
-    {ok, integer(), list(), pid()} |
+    {ok, integer(), list(), binary()} |
     {ok, integer(), list()} |
     {ok, pid()} |
     {error, term()}.
@@ -35,6 +40,91 @@ request(Method, Url, Headers, Body, Options) ->
 
     hackney:request(Method, Url , FinalHeaders, Body, FinalOpts).
 
+%% @doc Make a streaming GET/POST request and consume the initial status
+%% and headers, returning {ok, Status, Headers, Ref}. The caller can then
+%% pull the response body one chunk at a time with stream_body/1. This
+%% replaces the pull-based response streaming removed in hackney 4.x by
+%% driving hackney's async ({async, once}) flow-controlled mode.
+-spec stream_request(atom(), binary(), list(), term(), list()) ->
+    {ok, integer(), list(), pid()} | {error, term()}.
+stream_request(Method, Url, Headers, Body, Options) ->
+    {FinalHeaders, FinalOpts} = make_headers(Method, Url, Headers, Options),
+    Options1 = [{async, once} | FinalOpts],
+    case hackney:request(Method, Url, FinalHeaders, Body, Options1) of
+        {ok, Ref} ->
+            %% In {async, once} mode hackney delivers the status and headers
+            %% messages immediately, before waiting for stream_next/1.
+            receive
+                {hackney_response, Ref, {status, Status, _Reason}} ->
+                    receive
+                        {hackney_response, Ref, {headers, RespHeaders}} ->
+                            {ok, Status, RespHeaders, Ref};
+                        {hackney_response, Ref, {error, Reason}} ->
+                            {error, Reason}
+                    after ?DEFAULT_TIMEOUT ->
+                        {error, timeout}
+                    end;
+                {hackney_response, Ref, {error, Reason}} ->
+                    {error, Reason}
+            after ?DEFAULT_TIMEOUT ->
+                {error, timeout}
+            end;
+        Error ->
+            Error
+    end.
+
+%% @doc Pull the next response body chunk from a stream started with
+%% stream_request/5. Mirrors the old hackney:stream_body/1 contract:
+%% {ok, Data} | done | {error, term()}.
+-spec stream_body(pid()) -> {ok, binary()} | done | {error, term()}.
+stream_body(Ref) ->
+    hackney:stream_next(Ref),
+    receive
+        {hackney_response, Ref, done} ->
+            done;
+        {hackney_response, Ref, {error, Reason}} ->
+            {error, Reason};
+        {hackney_response, Ref, Data} when is_binary(Data) ->
+            {ok, Data};
+        %% tolerate late status/headers messages
+        {hackney_response, Ref, {status, _, _}} ->
+            stream_body(Ref);
+        {hackney_response, Ref, {headers, _}} ->
+            stream_body(Ref)
+    after ?DEFAULT_TIMEOUT ->
+        {error, timeout}
+    end.
+
+%% @doc Drain a stream started with stream_request/5 into a single binary.
+-spec stream_body_all(pid()) -> {ok, binary()} | {error, term()}.
+stream_body_all(Ref) ->
+    stream_body_all(Ref, []).
+
+stream_body_all(Ref, Acc) ->
+    case stream_body(Ref) of
+        {ok, Data} ->
+            stream_body_all(Ref, [Data | Acc]);
+        done ->
+            {ok, iolist_to_binary(lists:reverse(Acc))};
+        {error, _} = Error ->
+            Error
+    end.
+
+%% @doc Read the response after a streamed body upload (start_response/1)
+%% into the eager {ok, Status, Headers, Body} shape used by db_resp/2.
+-spec read_response(pid()) ->
+    {ok, integer(), list(), binary()} | {error, term()}.
+read_response(Ref) ->
+    case hackney:start_response(Ref) of
+        {ok, Status, Headers, ConnPid} ->
+            case hackney:body(ConnPid) of
+                {ok, Body} -> {ok, Status, Headers, Body};
+                {error, _} = Error -> Error
+            end;
+        Error ->
+            Error
+    end.
+
 db_request(Method, Url, Headers, Body, Options) ->
     db_request(Method, Url, Headers, Body, Options, []).
 
@@ -42,13 +132,10 @@ db_request(Method, Url, Headers, Body, Options, Expect) ->
     Resp = request(Method, Url, Headers, Body, Options),
     db_resp(Resp, Expect).
 
-json_body(Ref) ->
-    case hackney:body(Ref) of
-        {ok, Body} ->
-            couchbeam_ejson:decode(Body);
-        {error, _} = Error ->
-            Error
-    end.
+%% @doc Decode a JSON response body. With hackney 4.x request/5 reads the
+%% body eagerly into a binary, so this just decodes it.
+json_body(Body) when is_binary(Body) ->
+    couchbeam_ejson:decode(Body).
 
 make_headers(Method, Url, Headers, Options) ->
     Headers1 = case couchbeam_util:get_value(<<"Accept">>, Headers) of
@@ -78,7 +165,7 @@ maybe_proxyauth_header(Headers, Options) ->
       {lists:append([ProxyauthProps,Headers]), proplists:delete(proxyauth, Options)}
   end.
 
-db_resp({ok, Ref}=Resp, _Expect) when is_reference(Ref) ->
+db_resp({ok, Ref}=Resp, _Expect) when is_pid(Ref) orelse is_reference(Ref) ->
     Resp;
 db_resp({ok, 401, _}, _Expect) ->
     {error, unauthenticated};
@@ -98,40 +185,26 @@ db_resp({ok, Status, Headers}=Resp, Expect) ->
         false ->
             {error, {bad_response, {Status, Headers, <<>>}}}
     end;
-db_resp({ok, 401, _, Ref}, _Expect) ->
-    hackney:skip_body(Ref),
+db_resp({ok, 401, _, _}, _Expect) ->
     {error, unauthenticated};
-db_resp({ok, 403, _, Ref}, _Expect) ->
-    hackney:skip_body(Ref),
+db_resp({ok, 403, _, _}, _Expect) ->
     {error, forbidden};
-db_resp({ok, 404, _, Ref}, _Expect) ->
-    hackney:skip_body(Ref),
+db_resp({ok, 404, _, _}, _Expect) ->
     {error, not_found};
-db_resp({ok, 409, _, Ref}, _Expect) ->
-    hackney:skip_body(Ref),
+db_resp({ok, 409, _, _}, _Expect) ->
     {error, conflict};
-db_resp({ok, 412, _, Ref}, _Expect) ->
-    hackney:skip_body(Ref),
+db_resp({ok, 412, _, _}, _Expect) ->
     {error, precondition_failed};
 db_resp({ok, _, _, _}=Resp, []) ->
     Resp;
-db_resp({ok, Status, Headers, Ref}=Resp, Expect) ->
+db_resp({ok, Status, Headers, Body}=Resp, Expect) ->
     case lists:member(Status, Expect) of
         true -> Resp;
-        false -> {error, {bad_response, {Status, Headers, db_resp_body(Ref)}}}
+        false -> {error, {bad_response, {Status, Headers, Body}}}
     end;
 db_resp(Error, _Expect) ->
     Error.
 
-db_resp_body(Ref) ->
-    case hackney:body(Ref) of
-        {ok, Body} -> Body;
-        {error, _} ->
-            %% Try to close the connection to prevent leaks
-            catch hackney:close(Ref),
-            <<>>
-    end.
-
 -spec server_url(server()) -> binary().
 %% @doc Asemble the server URL for the given client
 server_url(#server{url=Url}) ->
@@ -153,21 +226,18 @@ reply_att(ok) ->
     ok;
 reply_att(done) ->
     done;
-reply_att({ok, 404, _, Ref}) ->
-    hackney:skip_body(Ref),
+reply_att({ok, 404, _, _}) ->
     {error, not_found};
-reply_att({ok, 409, _, Ref}) ->
-    hackney:skip_body(Ref),
+reply_att({ok, 409, _, _}) ->
     {error, conflict};
-reply_att({ok, Status, _, Ref}) when Status =:= 200 orelse Status =:= 201 ->
-  case couchbeam_httpc:json_body(Ref) of
+reply_att({ok, Status, _, Body}) when Status =:= 200 orelse Status =:= 201 ->
+  case couchbeam_httpc:json_body(Body) of
       #{<<"ok">> := true} = Map ->
           {ok, maps:remove(<<"ok">>, Map)};
-      {error, _} = Error ->
-          Error
+      Other ->
+          {error, {bad_response, Other}}
   end;
-reply_att({ok, Status, Headers, Ref}) ->
-    {ok, Body} = hackney:body(Ref),
+reply_att({ok, Status, Headers, Body}) ->
     {error, {bad_response, {Status, Headers, Body}}};
 reply_att(Error) ->
     Error.
@@ -186,7 +256,7 @@ wait_mp_doc_loop(Ref, Parser, InputBuffer, DocBuffer) ->
             wait_mp_doc_body(Ref, BodyCont, DocBuffer);
         {more, NewParser} ->
             %% Need more data from the connection
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, Data} ->
                     wait_mp_doc_loop(Ref, NewParser, Data, DocBuffer);
                 done ->
@@ -227,7 +297,7 @@ wait_mp_doc_body(Ref, BodyCont, DocBuffer) ->
             end;
         {more, MoreBodyCont} ->
             %% Need more data
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, Data} ->
                     %% Feed the data and continue
                     wait_mp_doc_body_with_data(Ref, MoreBodyCont, DocBuffer, Data);
@@ -276,7 +346,7 @@ wait_mp_doc_next_part(Ref, NextPartCont) ->
         {headers, _Headers, BodyCont} ->
             wait_mp_doc_body(Ref, BodyCont, <<>>);
         {more, Parser} ->
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, Data} ->
                     wait_mp_doc_loop(Ref, Parser, Data, <<>>);
                 done ->
@@ -314,7 +384,7 @@ wait_mp_att_next_part(Ref, NextPartCont, {AttName, AttNames}) ->
                     {att, Name, NState}
             end;
         {more, Parser} ->
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, Data} ->
                     wait_mp_att_parse(Ref, Parser, Data, {AttName, AttNames});
                 done ->
@@ -350,7 +420,7 @@ wait_mp_att_parse(Ref, Parser, Data, {AttName, AttNames}) ->
                     {att, Name, NState}
             end;
         {more, NewParser} ->
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, NewData} ->
                     wait_mp_att_parse(Ref, NewParser, NewData, {AttName, AttNames});
                 done ->
@@ -374,7 +444,7 @@ wait_mp_att_body(Ref, BodyCont, {AttName, AttNames}) ->
             NState = {Ref, fun() -> wait_mp_att_next_part(Ref, NextPartCont, {nil, AttNames}) end},
             {att_eof, AttName, NState};
         {more, MoreBodyCont} ->
-            case hackney:stream_body(Ref) of
+            case stream_body(Ref) of
                 {ok, Data} ->
                     wait_mp_att_body_with_data(Ref, MoreBodyCont, Data, {AttName, AttNames});
                 done ->
@@ -529,10 +599,10 @@ send_mp_doc_atts([Att | Rest], Ref, Doc, Boundary) ->
 
 %% @hidden
 mp_doc_reply(Ref, Doc) ->
-    Resp = hackney:start_response(Ref),
+    Resp = read_response(Ref),
     case couchbeam_httpc:db_resp(Resp, [200, 201]) of
-        {ok, _, _, Ref} ->
-            JsonProp = couchbeam_httpc:json_body(Ref),
+        {ok, _, _, Body} ->
+            JsonProp = couchbeam_httpc:json_body(Body),
             NewRev = maps:get(<<"rev">>, JsonProp),
             NewDocId = maps:get(<<"id">>, JsonProp),
             %% set the new doc ID
diff --git a/src/couchbeam_util.erl b/src/couchbeam_util.erl
index 865c4c6c..7fd46421 100644
--- a/src/couchbeam_util.erl
+++ b/src/couchbeam_util.erl
@@ -233,8 +233,8 @@ shutdown_sync(Pid) when not is_pid(Pid)->
 shutdown_sync(Pid) ->
     MRef = erlang:monitor(process, Pid),
     try
-        catch unlink(Pid),
-        catch exit(Pid, shutdown),
+        try unlink(Pid) catch _:_ -> ok end,
+        try exit(Pid, shutdown) catch _:_ -> ok end,
         receive
         {'DOWN', MRef, _, _, _} ->
             ok
diff --git a/src/couchbeam_uuids.erl b/src/couchbeam_uuids.erl
index 96da117e..037cab99 100644
--- a/src/couchbeam_uuids.erl
+++ b/src/couchbeam_uuids.erl
@@ -121,15 +121,13 @@ get_new_uuids(#server{url=ServerUrl, options=Opts}=Server, Backoff, Acc) ->
     Count = list_to_binary(integer_to_list(1000 - length(Acc))),
     Url = hackney_url:make_url(ServerUrl, <<"/_uuids">>, [{<<"count">>, Count}]),
     case couchbeam_httpc:request(get, Url, [], <<>>, Opts) of
-        {ok, 200, _, Ref} ->
-            {ok, Body} = hackney:body(Ref),
+        {ok, 200, _, Body} ->
             #{<<"uuids">> := Uuids} = couchbeam_ejson:decode(Body),
             ServerUuids = #server_uuids{server_url=ServerUrl,
                                         uuids=(Acc ++ Uuids)},
             ets:insert(couchbeam_uuids, ServerUuids),
             {ok, ServerUuids};
-        {ok, Status, Headers, Ref} ->
-            {ok, Body} = hackney:body(Ref),
+        {ok, Status, Headers, Body} ->
             {error, {bad_response, {Status, Headers, Body}}};
         {error, closed} ->
             wait_for_retry(Server, Backoff, Acc);
diff --git a/src/couchbeam_view.erl b/src/couchbeam_view.erl
index 60e433c6..d6b838b3 100644
--- a/src/couchbeam_view.erl
+++ b/src/couchbeam_view.erl
@@ -670,9 +670,8 @@ decode_view_data(Data, #view_stream_state{owner = Owner, ref = Ref,
     lists:foreach(fun(Row) -> Owner ! {Ref, {row, Row}} end, Rows),
     case Parser1 of
         #st{phase = done} ->
-            %% All rows parsed
-            catch hackney:stop_async(ClientRef),
-            catch hackney:skip_body(ClientRef),
+            %% All rows parsed - close the connection
+            try hackney:close(ClientRef) catch _:_ -> ok end,
             Owner ! {Ref, done};
         _ ->
             maybe_continue_view(State#view_stream_state{parser = Parser1})
@@ -780,7 +779,6 @@ view_notfound_test() ->
 %% Helper to generate mock view responses based on URL
 view_mock_response(Url) ->
     UrlBin = iolist_to_binary(Url),
-    Ref = make_ref(),
     %% Check for limit=1 (used by first/3)
     HasLimit1 = case binary:match(UrlBin, <<"limit=1">>) of
         nomatch -> false;
@@ -849,7 +847,6 @@ view_mock_response(Url) ->
                   #{<<"id">> => <<"doc2">>, <<"key">> => <<"doc2">>, <<"value">> => #{}}
               ]}
     end,
-    couchbeam_mocks:set_body(Ref, Response),
-    {ok, 200, [], Ref}.
+    {ok, 200, [], couchbeam_mocks:body(Response)}.
 
 -endif.
diff --git a/test/couchbeam_mocks.erl b/test/couchbeam_mocks.erl
index f33d09b4..458460ef 100644
--- a/test/couchbeam_mocks.erl
+++ b/test/couchbeam_mocks.erl
@@ -7,35 +7,24 @@
 
 -export([setup/0, teardown/0]).
 -export([expect_response/3, expect_db_response/3]).
--export([set_body/2, get_body/1]).
+-export([body/1]).
 
 %% @doc Initialize meck for hackney and couchbeam_httpc
 setup() ->
     %% Unload any existing mocks first
-    catch meck:unload(hackney),
-    catch meck:unload(couchbeam_httpc),
+    try meck:unload(hackney) catch _:_ -> ok end,
+    try meck:unload(couchbeam_httpc) catch _:_ -> ok end,
 
     %% Create mocks with passthrough for unmocked functions
     meck:new(hackney, [passthrough, no_link]),
     meck:new(couchbeam_httpc, [passthrough, no_link]),
 
-    %% Setup default mock for hackney:body to read from process dict
-    meck:expect(hackney, body, fun(Ref) ->
-        case get_body(Ref) of
-            undefined -> {error, not_found};
-            Body -> {ok, Body}
-        end
-    end),
-
-    %% Setup default mock for hackney:skip_body
-    meck:expect(hackney, skip_body, fun(_Ref) -> ok end),
-
     ok.
 
 %% @doc Unload all mocks
 teardown() ->
-    catch meck:unload(hackney),
-    catch meck:unload(couchbeam_httpc),
+    try meck:unload(hackney) catch _:_ -> ok end,
+    try meck:unload(couchbeam_httpc) catch _:_ -> ok end,
     ok.
 
 %% @doc Set up an expectation for hackney:request
@@ -45,15 +34,7 @@ expect_response(Matcher, Response, JsonBody) when is_function(Matcher, 5) ->
     meck:expect(hackney, request, fun(Method, Url, Headers, Body, Opts) ->
         case Matcher(Method, Url, Headers, Body, Opts) of
             true ->
-                case Response of
-                    {ok, Status, RespHeaders, Ref} ->
-                        set_body(Ref, JsonBody),
-                        {ok, Status, RespHeaders, Ref};
-                    {ok, Status, RespHeaders} ->
-                        {ok, Status, RespHeaders};
-                    Other ->
-                        Other
-                end;
+                mock_response(Response, JsonBody);
             false ->
                 meck:passthrough([Method, Url, Headers, Body, Opts])
         end
@@ -67,29 +48,23 @@ expect_db_response(Matcher, Response, JsonBody) when is_function(Matcher, 5) ->
         fun(Method, Url, Headers, Body, Opts, _Expect) ->
             case Matcher(Method, Url, Headers, Body, Opts) of
                 true ->
-                    case Response of
-                        {ok, Status, RespHeaders, Ref} ->
-                            set_body(Ref, JsonBody),
-                            {ok, Status, RespHeaders, Ref};
-                        {ok, Status, RespHeaders} ->
-                            {ok, Status, RespHeaders};
-                        Other ->
-                            Other
-                    end;
+                    mock_response(Response, JsonBody);
                 false ->
                     meck:passthrough([Method, Url, Headers, Body, Opts, _Expect])
             end
         end).
 
-%% @doc Store a response body for a given ref
-set_body(Ref, Body) when is_binary(Body) ->
-    put({mock_body, Ref}, Body);
-set_body(Ref, Term) ->
-    put({mock_body, Ref}, couchbeam_ejson:encode(Term)).
+%% @doc With hackney 4.x the body is returned eagerly in the response
+%% tuple, so embed it directly.
+mock_response({ok, Status, RespHeaders, _Ref}, JsonBody) ->
+    {ok, Status, RespHeaders, body(JsonBody)};
+mock_response({ok, Status, RespHeaders}, _JsonBody) ->
+    {ok, Status, RespHeaders};
+mock_response(Other, _JsonBody) ->
+    Other.
 
-%% @doc Retrieve and erase a stored response body
-get_body(Ref) ->
-    case erase({mock_body, Ref}) of
-        undefined -> undefined;
-        Body -> Body
-    end.
+%% @doc Encode a term (or pass a binary through) as a response body binary.
+body(Body) when is_binary(Body) ->
+    Body;
+body(Term) ->
+    couchbeam_ejson:encode(Term).

From 32034ce28a4146b2ff1a65a1e3a39f6e0621453c Mon Sep 17 00:00:00 2001
From: Benoit Chesneau <bchesneau@gmail.com>
Date: Tue, 9 Jun 2026 14:40:03 +0200
Subject: [PATCH 2/3] fix CI: once-mode view streaming and FreeBSD deps

- couchbeam_view: gate {async, once} streaming on the owner's stream_next
  so the stream process stays alive until the first batch is consumed
  (it was racing to exit, which broke stream_next/1)
- integration suite: drop removed hackney:skip_body/1 in create_mango_index
- FreeBSD CI: pkg upgrade before install to fix the git/pcre2 ABI mismatch
---
 .github/workflows/test.yml           |  3 +++
 src/couchbeam_view.erl               | 34 ++++++++++++++++++++++------
 test/couchbeam_integration_SUITE.erl |  3 +--
 3 files changed, 31 insertions(+), 9 deletions(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index 490332ee..ed5c7fe8 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -102,6 +102,9 @@ jobs:
         release: "14.2"
         usesh: true
         prepare: |
+          # Upgrade base packages first so git matches the current pcre2 ABI
+          pkg update -f
+          pkg upgrade -y
           pkg install -y erlang-runtime28 rebar3 cmake git gmake go llvm18
         run: |
           # Ensure Erlang 28 is in PATH
diff --git a/src/couchbeam_view.erl b/src/couchbeam_view.erl
index d6b838b3..0c38934d 100644
--- a/src/couchbeam_view.erl
+++ b/src/couchbeam_view.erl
@@ -663,19 +663,39 @@ view_stream_loop(#view_stream_state{owner = Owner, ref = Ref, mref = MRef,
     end.
 
 decode_view_data(Data, #view_stream_state{owner = Owner, ref = Ref,
-                                           client_ref = ClientRef,
                                            parser = Parser} = State) ->
     {Rows, Parser1} = json_stream_parse:feed(Data, Parser),
     %% Send each row to owner
     lists:foreach(fun(Row) -> Owner ! {Ref, {row, Row}} end, Rows),
-    case Parser1 of
-        #st{phase = done} ->
-            %% All rows parsed - close the connection
+    Done = case Parser1 of #st{phase = done} -> true; _ -> false end,
+    continue_view(State#view_stream_state{parser = Parser1}, Done).
+
+%% In {async, once} mode, wait for the owner to request the next batch before
+%% pulling more from the connection or finishing. This keeps the stream
+%% process alive until the owner has consumed the first batch (and can call
+%% stream_next/1). After the first request we drain the rest without further
+%% gating, so the owner receives all remaining rows and the final done.
+continue_view(#view_stream_state{ref = Ref, mref = MRef, owner = Owner,
+                                 client_ref = ClientRef,
+                                 async = once} = State, Done) ->
+    receive
+        {'DOWN', MRef, _, _, _} ->
+            exit(normal);
+        {Ref, cancel} ->
+            try hackney:close(ClientRef) catch _:_ -> ok end,
+            Owner ! {Ref, ok};
+        {Ref, stream_next} when Done ->
             try hackney:close(ClientRef) catch _:_ -> ok end,
             Owner ! {Ref, done};
-        _ ->
-            maybe_continue_view(State#view_stream_state{parser = Parser1})
-    end.
+        {Ref, stream_next} ->
+            view_stream_loop(State#view_stream_state{async = normal})
+    end;
+continue_view(#view_stream_state{owner = Owner, ref = Ref,
+                                 client_ref = ClientRef}, true) ->
+    try hackney:close(ClientRef) catch _:_ -> ok end,
+    Owner ! {Ref, done};
+continue_view(State, false) ->
+    maybe_continue_view(State).
 
 maybe_continue_view(#view_stream_state{ref = Ref, mref = MRef, owner = Owner,
                                         client_ref = ClientRef,
diff --git a/test/couchbeam_integration_SUITE.erl b/test/couchbeam_integration_SUITE.erl
index 2e939936..d821de78 100644
--- a/test/couchbeam_integration_SUITE.erl
+++ b/test/couchbeam_integration_SUITE.erl
@@ -1962,8 +1962,7 @@ create_mango_index(#db{server=Server, options=Opts}=Db, Fields) ->
                {<<"accept">>, <<"application/json">>}],
     Body = couchbeam_ejson:encode(IndexSpec),
     case couchbeam_httpc:db_request(post, Url, Headers, Body, Opts, [200, 201]) of
-        {ok, _, _, Ref} ->
-            hackney:skip_body(Ref),
+        {ok, _, _, _} ->
             ct:pal("Created Mango index on fields: ~p", [Fields]),
             ok;
         {error, Reason} ->

From 929280f1d5115949ffb1bbe91c8feea02a0453b9 Mon Sep 17 00:00:00 2001
From: Benoit Chesneau <bchesneau@gmail.com>
Date: Tue, 9 Jun 2026 14:44:05 +0200
Subject: [PATCH 3/3] fix FreeBSD CI: ignore pkg OS-version mismatch (repo is
 on a newer minor)

---
 .github/workflows/test.yml | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index ed5c7fe8..953ba3d3 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -102,7 +102,10 @@ jobs:
         release: "14.2"
         usesh: true
         prepare: |
-          # Upgrade base packages first so git matches the current pcre2 ABI
+          # The pkg repo is built for a newer FreeBSD minor than the VM image,
+          # so allow the OS-version mismatch and upgrade installed packages
+          # first to keep ABIs (e.g. git/pcre2) consistent.
+          export IGNORE_OSVERSION=yes
           pkg update -f
           pkg upgrade -y
           pkg install -y erlang-runtime28 rebar3 cmake git gmake go llvm18