(couchdb) branch also-fix-sws-for-plain-attachment-uploads created (now 3bbbea501)

[va...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nickva pushed a change to branch also-fix-sws-for-plain-attachment-uploads
in repository https://gitbox.apache.org/repos/asf/couchdb.git


      at 3bbbea501 Fix all stream attachments for serialize_worker_startup

This branch includes the following new commits:

     new 3bbbea501 Fix all stream attachments for serialize_worker_startup

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(couchdb) 01/01: Fix all stream attachments for serialize_worker_startup

[va...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

nickva pushed a commit to branch also-fix-sws-for-plain-attachment-uploads
in repository https://gitbox.apache.org/repos/asf/couchdb.git

commit 3bbbea5010ac05a3c996f517cbcc040ed819d125
Author: Nick Vatamaniuc <va...@gmail.com>
AuthorDate: Wed Jul 22 16:24:49 2026 -0400

    Fix all stream attachments for serialize_worker_startup
    
    In #6071 we fixed multipart attachment uploads but we forgot to fix plain
    attachment uploads with `PUT /db/doc/attachment`. We should apply the same fix
    for these too, otherwise they'll also buffer whole attachment in memory when
    sws=true, defeating the whole purpose of incremental attachment streaming.
    
    And add a few parsing tests, especially testing how workers wait on others and
    how the parser should behave if the attachment is too short
---
 .../test/eunit/couch_httpd_multipart_tests.erl     | 148 +++++++++++++++++++++
 src/fabric/src/fabric_doc_update.erl               |  51 +++++--
 2 files changed, 185 insertions(+), 14 deletions(-)

diff --git a/src/couch/test/eunit/couch_httpd_multipart_tests.erl b/src/couch/test/eunit/couch_httpd_multipart_tests.erl
new file mode 100644
index 000000000..7411f7611
--- /dev/null
+++ b/src/couch/test/eunit/couch_httpd_multipart_tests.erl
@@ -0,0 +1,148 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+%   http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(couch_httpd_multipart_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+
+-define(CONTENT_TYPE, ~S'multipart/related;boundary="abc123"').
+-define(DOC_JSON,
+    ~B'{"_attachments":{"ohai":{"follows":true,"content_type":"text/plain","length":4}}}'
+).
+-define(ATT_LEN, 4).
+
+couch_httpd_multipart_test_() ->
+    {
+        foreach,
+        fun setup/0,
+        fun teardown/1,
+        [
+            ?TDEF_FE(t_single_writer_short_part_fails_fast),
+            ?TDEF_FE(t_single_writer_empty_part_fails_fast),
+            ?TDEF_FE(t_single_writer_exact_part_waits_for_more_writers),
+            ?TDEF_FE(t_all_writers_short_part_fail)
+        ]
+    }.
+
+setup() ->
+    test_util:start_applications([config]).
+
+teardown(Apps) ->
+    test_util:stop_applications(Apps).
+
+% Doc with on attachment declaring length 4 but only having length 3. The
+% writer which detects doesn't have enough bytes exits.
+t_single_writer_short_part_fails_fast(_) ->
+    {Parser, ParserRef, Ref} = parse(3, body(<<"oha">>)),
+    Writer = spawn_writer(Parser, Ref, ?ATT_LEN),
+    ?assertEqual({parser_died, normal}, wait_writer(Writer, 3000)),
+    ?assertEqual(normal, wait_down(ParserRef, 3000)),
+    stop_writers([Writer]).
+
+% Empty body part. Writer won't get any chunk before it starves.
+t_single_writer_empty_part_fails_fast(_) ->
+    {Parser, ParserRef, Ref} = parse(3, body(<<>>)),
+    Writer = spawn_writer(Parser, Ref, ?ATT_LEN),
+    ?assertEqual({parser_died, normal}, wait_writer(Writer, 3000)),
+    ?assertEqual(normal, wait_down(ParserRef, 3000)),
+    stop_writers([Writer]).
+
+% First writer consuming all bytes should not exit and wait for other to also
+% connect and fetch their copies.
+t_single_writer_exact_part_waits_for_more_writers(_) ->
+    {Parser, ParserRef, Ref} = parse(3, body(<<"ohai">>)),
+    Writer1 = spawn_writer(Parser, Ref, ?ATT_LEN),
+    ?assertEqual(got_all_bytes, wait_writer(Writer1, 3000)),
+    % Check it didn't die
+    ?assertEqual(timeout, wait_down(ParserRef, 300)),
+    % Late-arriving writers
+    Writer2 = spawn_writer(Parser, Ref, ?ATT_LEN),
+    ?assertEqual(got_all_bytes, wait_writer(Writer2, 3000)),
+    % Still shouldn't die. One more left
+    ?assertEqual(timeout, wait_down(ParserRef, 300)),
+    Writer3 = spawn_writer(Parser, Ref, ?ATT_LEN),
+    ?assertEqual(got_all_bytes, wait_writer(Writer3, 3000)),
+    % Everyone got their bytes, now it can die
+    ?assertEqual(normal, wait_down(ParserRef, 3000)),
+    stop_writers([Writer1, Writer2, Writer3]).
+
+% Multuple writers get the "too short" condition. That's fine, everyone dies
+% without blocking.
+t_all_writers_short_part_fail(_) ->
+    {Parser, ParserRef, Ref} = parse(3, body(<<"oha">>)),
+    Writers = [spawn_writer(Parser, Ref, ?ATT_LEN) || _ <- lists:seq(1, 3)],
+    [?assertEqual({parser_died, normal}, wait_writer(W, 3000)) || W <- Writers],
+    ?assertEqual(normal, wait_down(ParserRef, 3000)),
+    stop_writers(Writers).
+
+body(AttBytes) ->
+    <<
+        "--abc123\r\n"
+        "Content-Type: application/json\r\n"
+        "\r\n",
+        ?DOC_JSON/binary,
+        "\r\n"
+        "--abc123\r\n"
+        "\r\n",
+        AttBytes/binary,
+        "\r\n"
+        "--abc123--"
+    >>.
+
+% Spawn the parser like chttpd_db for MP PUT
+parse(NumWriters, Body) ->
+    Ref = make_ref(),
+    couch_httpd_multipart:num_mp_writers(NumWriters),
+    DataFun = fun() -> {Body, fun() -> throw(<<"expected more data">>) end} end,
+    {{doc_bytes, Ref, DocBytes}, Parser, ParserRef} =
+        couch_httpd_multipart:decode_multipart_stream(?CONTENT_TYPE, DataFun, Ref),
+    ?assertEqual(?DOC_JSON, iolist_to_binary(DocBytes)),
+    {Parser, ParserRef, Ref}.
+
+% This is like the reader loop in fabric_rpc:make_att_reader()
+spawn_writer(Parser, Ref, Need) ->
+    ReportTo = self(),
+    spawn_link(fun() ->
+        ParserRef = monitor(process, Parser),
+        Parser ! {hello_from_writer, Ref, self()},
+        ReportTo ! {self(), writer_read(Parser, ParserRef, Ref, Need)},
+        receive
+            stop -> ok
+        end
+    end).
+
+writer_read(_Parser, _ParserRef, _Ref, Need) when Need =< 0 ->
+    got_all_bytes;
+writer_read(Parser, ParserRef, Ref, Need) ->
+    Parser ! {get_bytes, Ref, self()},
+    receive
+        {bytes, Ref, Bytes} ->
+            writer_read(Parser, ParserRef, Ref, Need - iolist_size(Bytes));
+        {'DOWN', ParserRef, _, _, Reason} ->
+            {parser_died, Reason}
+    end.
+
+wait_writer(Writer, Timeout) ->
+    receive
+        {Writer, Result} -> Result
+    after Timeout -> timeout
+    end.
+
+wait_down(MonitorRef, Timeout) ->
+    receive
+        {'DOWN', MonitorRef, _, _, Reason} -> Reason
+    after Timeout -> timeout
+    end.
+
+stop_writers(Writers) ->
+    [Writer ! stop || Writer <- Writers],
+    ok.
diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl
index f1199c6a7..a73e09353 100644
--- a/src/fabric/src/fabric_doc_update.erl
+++ b/src/fabric/src/fabric_doc_update.erl
@@ -434,24 +434,26 @@ validate_atomic_update(_DbName, AllDocs, true) ->
     ),
     throw({aborted, PreCommitFailures}).
 
-% Replicated changes and multipart attachment are always in parallel. MP parser
-% is designed to distribute attachment chunks to num_mp_writers concurrently,
-% so if we serialize them we make the MP parser buffer all the attachment
-% chunks (say 1GB of data) before the subsequent workers will be started.
+% Replicated changes streamed attachments are always in parallel. Both MP
+% parser and fabric_doc_atts are designed to distribute attachment chunks
+% concurrently. If we serialize them they would always buffer the whole
+% attachment in memory (say 1GB of data) until all workers have consumed the
+% byte ranges.
 serialize_worker_startup(AllDocs, Options) ->
     Replicated = proplists:get_value(?REPLICATED_CHANGES, Options) =:= true,
-    case Replicated orelse any_multipart_atts(AllDocs) of
+    case Replicated orelse any_streamed_atts(AllDocs) of
         true -> false;
         false -> config:get_boolean("fabric", "serialize_worker_startup", true)
     end.
 
-any_multipart_atts(Docs) ->
-    DocHasMpAtt = fun(#doc{atts = Atts}) -> lists:any(fun is_multipart_att/1, Atts) end,
-    lists:any(DocHasMpAtt, Docs).
+any_streamed_atts(Docs) ->
+    HasStreamedAtt = fun(#doc{atts = Atts}) -> lists:any(fun is_streamed_att/1, Atts) end,
+    lists:any(HasStreamedAtt, Docs).
 
-is_multipart_att(Att) ->
+is_streamed_att(Att) ->
     case couch_att:fetch(data, Att) of
         {follows, Parser, Ref} when is_pid(Parser), is_reference(Ref) -> true;
+        {fabric_attachment_receiver, Middleman, _} when is_pid(Middleman) -> true;
         _ -> false
     end.
 
@@ -561,7 +563,7 @@ doc_update_test_() ->
             fun filter_conflicts_drops_seen_docs/0,
             fun parallel_in_flight_after_conflict/0,
             fun serial_filters_conflicts_at_cast/0,
-            fun sws_multipart_atts_check/0,
+            fun sws_streamed_atts_check/0,
             fun sws_false_mode_conflict_not_final/0,
             fun sws_false_mode_ok_can_outvote_conflict/0,
             fun group_docs_content_and_order/0,
@@ -1174,8 +1176,10 @@ serial_filters_conflicts_at_cast() ->
     ?assertEqual([Tagged2], Stored).
 
 % Docs with streaming attachment don't serialize workers
-sws_multipart_atts_check() ->
+sws_streamed_atts_check() ->
     MpDoc = #doc{id = <<"m">>, atts = [mp_att()]},
+    ReceiverDoc = #doc{id = <<"r">>, atts = [receiver_att()]},
+    ChunkedReceiverDoc = #doc{id = <<"cr">>, atts = [chunked_receiver_att()]},
     InlineAtt = couch_att:new([
         {name, <<"b">>}, {type, <<"text/plain">>}, {att_len, 1}, {data, <<"x">>}
     ]),
@@ -1184,12 +1188,15 @@ sws_multipart_atts_check() ->
     ]),
     InlineDoc = #doc{id = <<"i">>, atts = [InlineAtt, StubAtt]},
     PlainDoc = #doc{id = <<"p">>},
-    % Without multipart atts the config default applies
+    % Without streamed atts the config default applies
     ?assert(serialize_worker_startup([PlainDoc], [])),
     ?assert(serialize_worker_startup([InlineDoc], [])),
-    % Any doc with a mp att forces parallel startup
+    % Parallel for mp parser and fabric attachment receiver
     ?assertNot(serialize_worker_startup([MpDoc], [])),
-    ?assertNot(serialize_worker_startup([PlainDoc, MpDoc], [])).
+    ?assertNot(serialize_worker_startup([PlainDoc, MpDoc], [])),
+    ?assertNot(serialize_worker_startup([ReceiverDoc], [])),
+    ?assertNot(serialize_worker_startup([PlainDoc, ReceiverDoc], [])),
+    ?assertNot(serialize_worker_startup([ChunkedReceiverDoc], [])).
 
 mp_att() ->
     couch_att:new([
@@ -1199,6 +1206,22 @@ mp_att() ->
         {data, {follows, self(), make_ref()}}
     ]).
 
+receiver_att() ->
+    couch_att:new([
+        {name, <<"a">>},
+        {type, <<"text/plain">>},
+        {att_len, 4},
+        {data, {fabric_attqachment_receiver, self(), 4}}
+    ]).
+
+chunked_receiver_att() ->
+    couch_att:new([
+        {name, <<"a">>},
+        {type, <<"text/plain">>},
+        {att_len, undefined},
+        {data, {fabric_attachment_receiver, self(), chunked}}
+    ]).
+
 sws_false_mode_conflict_not_final() ->
     Docs =
         [Doc1, Doc2] = tag_docs([