Make sure to wait for the attachments to uploaded before responding to the user

If workers drop after when new_edits=false and revisions are already present,
attachment parser will go into `mp_abort_parse_atts` state but still keep
consuming the request data, draining the uploaded attachment data, which helps
ensure the connection process will be in a consistent state and ready to
process the next HTTP request.

Add a test to PUT the same attachment multiple times with new_edits=false and
check that those requests all succeed.

2 files changed, 31 insertions, 21 deletions

diff --git a/src/chttpd/src/chttpd_db.erl b/src/chttpd/src/chttpd_db.erl
index 9c9c4ef87..4392df194 100644
--- a/src/chttpd/src/chttpd_db.erl
+++ b/src/chttpd/src/chttpd_db.erl
@@ -1206,9 +1206,16 @@ db_doc_req(#httpd{method = 'PUT', user_ctx = Ctx} = Req, Db, DocId) ->
             ),
             Doc = couch_doc_from_req(Req, Db, DocId, Doc0),
             try
-                Result = send_updated_doc(Req, Db, DocId, Doc, RespHeaders, UpdateType),
+                {HttpCode, RespHeaders1, RespBody} = update_doc_req(
+                    Req,
+                    Db,
+                    DocId,
+                    Doc,
+                    RespHeaders,
+                    UpdateType
+                ),
                 WaitFun(),
-                Result
+                send_json(Req, HttpCode, RespHeaders1, RespBody)
             catch
                 throw:Err ->
                     % Document rejected by a validate_doc_update function.
@@ -1534,14 +1541,12 @@ send_updated_doc(Req, Db, DocId, Json) ->
 send_updated_doc(Req, Db, DocId, Doc, Headers) ->
     send_updated_doc(Req, Db, DocId, Doc, Headers, interactive_edit).
 
-send_updated_doc(
+send_updated_doc(Req, Db, DocId, Doc, Headers, Type) ->
+    {Code, Headers1, Body} = update_doc_req(Req, Db, DocId, Doc, Headers, Type),
+    send_json(Req, Code, Headers1, Body).
+
+update_doc_req(Req, Db, DocId, Doc, Headers, UpdateType) ->
     #httpd{user_ctx = Ctx} = Req,
-    Db,
-    DocId,
-    #doc{deleted = Deleted} = Doc,
-    Headers,
-    UpdateType
-) ->
     W = chttpd:qs_value(Req, "w", integer_to_list(mem3:quorum(Db))),
     Options =
         case couch_httpd:header_value(Req, "X-Couch-Full-Commit") of
@@ -1552,15 +1557,10 @@ send_updated_doc(
             _ ->
                 [UpdateType, {user_ctx, Ctx}, {w, W}]
         end,
-    {Status, {etag, Etag}, Body} = update_doc(
-        Db,
-        DocId,
-        #doc{deleted = Deleted} = Doc,
-        Options
-    ),
+    {Status, {etag, Etag}, Body} = update_doc(Db, DocId, Doc, Options),
     HttpCode = http_code_from_status(Status),
     ResponseHeaders = [{"ETag", Etag} | Headers],
-    send_json(Req, HttpCode, ResponseHeaders, Body).
+    {HttpCode, ResponseHeaders, Body}.
 
 http_code_from_status(Status) ->
     case Status of
diff --git a/test/elixir/test/attachments_multipart_test.exs b/test/elixir/test/attachments_multipart_test.exs
index c5cd8fcdb..2cedef513 100644
--- a/test/elixir/test/attachments_multipart_test.exs
+++ b/test/elixir/test/attachments_multipart_test.exs
@@ -264,6 +264,8 @@ defmodule AttachmentMultipartTest do
   test "multipart attachments with new_edits=false", context do
     db_name = context[:db_name]
 
+    att_data = String.duplicate("x", 100_000)
+    att_len = byte_size(att_data)
     document = """
     {
       "body": "This is a body.",
@@ -271,7 +273,7 @@ defmodule AttachmentMultipartTest do
         "foo.txt": {
           "follows": true,
           "content_type": "application/test",
-          "length": 21
+          "length": #{att_len}
         }
       }
     }
@@ -284,7 +286,7 @@ defmodule AttachmentMultipartTest do
         document <>
         "\r\n--abc123\r\n" <>
         "\r\n" <>
-        "this is 21 chars long" <>
+        att_data <>
         "\r\n--abc123--epilogue"
 
     resp =
@@ -301,20 +303,28 @@ defmodule AttachmentMultipartTest do
 
     resp = Couch.get("/#{db_name}/multipart_replicated_changes/foo.txt")
 
-    assert resp.body == "this is 21 chars long"
+    assert resp.body == att_data
 
     # https://github.com/apache/couchdb/issues/3939
     # Repeating the request should not hang
+    Enum.each(0..10, fn _ ->
+      put_multipart_new_edits_false(db_name, rev, multipart_data)
+    end)
+  end
+
+  defp put_multipart_new_edits_false(db_name, rev, multipart_data) do
+    # Help ensure we're re-using client connections
+    ibrowse_opts = [{:max_sessions, 1}, {:max_pipeline_size, 1}]
     resp =
       Couch.put(
         "/#{db_name}/multipart_replicated_changes?new_edits=false&rev=#{rev}",
         body: multipart_data,
-        headers: ["Content-Type": "multipart/related;boundary=\"abc123\""]
+        headers: ["Content-Type": "multipart/related;boundary=\"abc123\""],
+        ibrowse: ibrowse_opts
       )
 
     assert resp.status_code in [201, 202]
     assert resp.body["ok"] == true
-
   end
 
   defp test_multipart_att_compression(dbname) do