Implement attachment size limitsissue-769

Currently CouchDB has configurable single document body size limits, as well as
http request body limits, and this commit implements attachment size limit.

Maximum attachment size can be configured with:

```
[couchdb]

max_attachment_size = Bytes | infinity
```

`infinity` (i.e. no maximum) is the default value it also preserves the current
behavior.

Fixes #769

12 files changed, 526 insertions, 20 deletions

diff --git a/rel/overlay/etc/default.ini b/rel/overlay/etc/default.ini
index 30b2efa5c..fa1124d3d 100644
--- a/rel/overlay/etc/default.ini
+++ b/rel/overlay/etc/default.ini
@@ -43,6 +43,8 @@ changes_doc_ids_optimization_threshold = 100
 ; Single documents that exceed this value in a bulk request will receive a
 ; too_large error. The max_http_request_size still takes precedence.
 ;single_max_doc_size = 1048576
+; Maximum attachment size.
+; max_attachment_size = infinity
 
 [cluster]
 q=8
diff --git a/src/chttpd/src/chttpd.erl b/src/chttpd/src/chttpd.erl
index 425f95a6f..6be0d1848 100644
--- a/src/chttpd/src/chttpd.erl
+++ b/src/chttpd/src/chttpd.erl
@@ -897,6 +897,8 @@ error_info({missing_stub, Reason}) ->
     {412, <<"missing_stub">>, Reason};
 error_info(request_entity_too_large) ->
     {413, <<"too_large">>, <<"the request entity is too large">>};
+error_info({request_entity_too_large, {attachment, AttName}}) ->
+    {413, <<"attachment_too_large">>, AttName};
 error_info({request_entity_too_large, DocID}) ->
     {413, <<"document_too_large">>, DocID};
 error_info({error, security_migration_updates_disabled}) ->
diff --git a/src/chttpd/test/chttpd_db_attachment_size_tests.erl b/src/chttpd/test/chttpd_db_attachment_size_tests.erl
new file mode 100644
index 000000000..0ab08dd80
--- /dev/null
+++ b/src/chttpd/test/chttpd_db_attachment_size_tests.erl
@@ -0,0 +1,206 @@
+% 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(chttpd_db_attachment_size_tests).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+
+-define(USER, "chttpd_db_att_test_admin").
+-define(PASS, "pass").
+-define(AUTH, {basic_auth, {?USER, ?PASS}}).
+-define(CONTENT_JSON, {"Content-Type", "application/json"}).
+-define(CONTENT_MULTI_RELATED, {"Content-Type",
+    "multipart/related;boundary=\"bound\""}).
+
+
+setup() ->
+    Hashed = couch_passwords:hash_admin_password(?PASS),
+    ok = config:set("admins", ?USER, ?b2l(Hashed), _Persist=false),
+    ok = config:set("couchdb", "max_attachment_size", "50", _Persist=false),
+    TmpDb = ?tempdb(),
+    Addr = config:get("chttpd", "bind_address", "127.0.0.1"),
+    Port = integer_to_list(mochiweb_socket_server:get(chttpd, port)),
+    Url = "http://" ++ Addr ++ ":" ++ Port ++ "/" ++ ?b2l(TmpDb),
+    create_db(Url),
+    add_doc(Url, "doc1"),
+    Url.
+
+
+teardown(Url) ->
+    delete_db(Url),
+    ok = config:delete("admins", ?USER, _Persist=false),
+    ok = config:delete("couchdb", "max_attachment_size").
+
+
+attachment_size_test_() ->
+    {
+        "chttpd max_attachment_size tests",
+        {
+            setup,
+            fun chttpd_test_util:start_couch/0,
+            fun chttpd_test_util:stop_couch/1,
+            {
+                foreach,
+                fun setup/0, fun teardown/1,
+                [
+                    fun put_inline/1,
+                    fun put_simple/1,
+                    fun put_simple_chunked/1,
+                    fun put_mp_related/1
+                ]
+            }
+        }
+    }.
+
+
+put_inline(Url) ->
+  ?_test(begin
+      Status = put_inline(Url, "doc2", 50),
+      ?assert(Status =:= 201 orelse Status =:= 202),
+      ?assertEqual(413, put_inline(Url, "doc3", 51))
+  end).
+
+
+put_simple(Url) ->
+    ?_test(begin
+        Headers = [{"Content-Type", "app/binary"}],
+        Rev1 = doc_rev(Url, "doc1"),
+        Data1 = data(50),
+        Status1 = put_req(Url ++ "/doc1/att2?rev=" ++ Rev1, Headers, Data1),
+        ?assert(Status1 =:= 201 orelse Status1 =:= 202),
+        Data2 = data(51),
+        Rev2 = doc_rev(Url, "doc1"),
+        Status2 = put_req(Url ++ "/doc1/att3?rev=" ++ Rev2, Headers, Data2),
+        ?assertEqual(413, Status2)
+    end).
+
+
+put_simple_chunked(Url) ->
+     ?_test(begin
+        Headers = [{"Content-Type", "app/binary"}],
+        Rev1 = doc_rev(Url, "doc1"),
+        DataFun1 = data_stream_fun(50),
+        Status1 = put_req_chunked(Url ++ "/doc1/att2?rev=" ++ Rev1, Headers, DataFun1),
+        ?assert(Status1 =:= 201 orelse Status1 =:= 202),
+        DataFun2 = data_stream_fun(51),
+        Rev2 = doc_rev(Url, "doc1"),
+        Status2 = put_req_chunked(Url ++ "/doc1/att3?rev=" ++ Rev2, Headers, DataFun2),
+        ?assertEqual(413, Status2)
+    end).
+
+
+put_mp_related(Url) ->
+    ?_test(begin
+        Headers = [?CONTENT_MULTI_RELATED],
+        Body1 = mp_body(50),
+        Status1 = put_req(Url ++ "/doc2", Headers, Body1),
+        ?assert(Status1 =:= 201 orelse Status1 =:= 202),
+        Body2 = mp_body(51),
+        Status2 = put_req(Url ++ "/doc3", Headers, Body2),
+        ?assertEqual(413, Status2)
+    end).
+
+
+% Helper functions
+
+create_db(Url) ->
+    Status = put_req(Url, "{}"),
+    ?assert(Status =:= 201 orelse Status =:= 202).
+
+
+add_doc(Url, DocId) ->
+    Status = put_req(Url ++ "/" ++ DocId, "{}"),
+    ?assert(Status =:= 201 orelse Status =:= 202).
+
+
+delete_db(Url) ->
+    {ok, 200, _, _} = test_request:delete(Url, [?AUTH]).
+
+
+put_inline(Url, DocId, Size) ->
+    Doc = "{\"_attachments\": {\"att1\":{"
+        "\"content_type\": \"app/binary\", "
+        "\"data\": \"" ++ data_b64(Size) ++ "\""
+        "}}}",
+    put_req(Url ++ "/" ++ DocId, Doc).
+
+
+mp_body(AttSize) ->
+    AttData = data(AttSize),
+    SizeStr = integer_to_list(AttSize),
+    string:join([
+        "--bound",
+
+        "Content-Type: application/json",
+
+        "",
+
+        "{\"_id\":\"doc2\", \"_attachments\":{\"att\":"
+        "{\"content_type\":\"app/binary\", \"length\":" ++ SizeStr ++ ","
+        "\"follows\":true}}}",
+
+        "--bound",
+
+        "Content-Disposition: attachment; filename=\"att\"",
+
+        "Content-Type: app/binary",
+
+        "",
+
+        AttData,
+
+        "--bound--"
+    ], "\r\n").
+
+
+doc_rev(Url, DocId) ->
+    {200, ResultProps} = get_req(Url ++ "/" ++ DocId),
+    {<<"_rev">>, BinRev} = lists:keyfind(<<"_rev">>, 1, ResultProps),
+    binary_to_list(BinRev).
+
+
+put_req(Url, Body) ->
+    put_req(Url, [], Body).
+
+
+put_req(Url, Headers, Body) ->
+    {ok, Status, _, _} = test_request:put(Url, Headers ++ [?AUTH], Body),
+    Status.
+
+
+put_req_chunked(Url, Headers, Body) ->
+    Opts = [{transfer_encoding, {chunked, 1}}],
+    {ok, Status, _, _} = test_request:put(Url, Headers ++ [?AUTH], Body, Opts),
+    Status.
+
+
+get_req(Url) ->
+    {ok, Status, _, ResultBody} = test_request:get(Url, [?CONTENT_JSON, ?AUTH]),
+    {[_ | _] = ResultProps} = ?JSON_DECODE(ResultBody),
+    {Status, ResultProps}.
+
+% Data streaming generator for ibrowse client. ibrowse will repeatedly call the
+% function with State and it should return {ok, Data, NewState} or eof at end.
+data_stream_fun(Size) ->
+    Fun = fun(0) -> eof; (BytesLeft) ->
+        {ok, <<"x">>, BytesLeft - 1}
+    end,
+    {Fun, Size}.
+
+
+data(Size) ->
+    string:copies("x", Size).
+
+
+data_b64(Size) ->
+    base64:encode_to_string(data(Size)).
diff --git a/src/couch/src/couch_att.erl b/src/couch/src/couch_att.erl
index 9d38cfae2..e78d6ef11 100644
--- a/src/couch/src/couch_att.erl
+++ b/src/couch/src/couch_att.erl
@@ -50,6 +50,11 @@
     downgrade/1
 ]).
 
+-export([
+    max_attachment_size/0,
+    validate_attachment_size/3
+]).
+
 -compile(nowarn_deprecated_type).
 -export_type([att/0]).
 
@@ -500,6 +505,8 @@ flush_data(Fd, Data, Att) when is_binary(Data) ->
         couch_stream:write(OutputStream, Data)
     end);
 flush_data(Fd, Fun, Att) when is_function(Fun) ->
+    AttName = fetch(name, Att),
+    MaxAttSize = max_attachment_size(),
     case fetch(att_len, Att) of
         undefined ->
             couch_db:with_stream(Fd, Att, fun(OutputStream) ->
@@ -510,7 +517,7 @@ flush_data(Fd, Fun, Att) when is_function(Fun) ->
                     % WriterFun({0, _Footers}, State)
                     % Called with Length == 0 on the last time.
                     % WriterFun returns NewState.
-                    fun({0, Footers}, _) ->
+                    fun({0, Footers}, _Total) ->
                         F = mochiweb_headers:from_binary(Footers),
                         case mochiweb_headers:get_value("Content-MD5", F) of
                         undefined ->
@@ -518,11 +525,15 @@ flush_data(Fd, Fun, Att) when is_function(Fun) ->
                         Md5 ->
                             {md5, base64:decode(Md5)}
                         end;
-                    ({_Length, Chunk}, _) ->
-                        couch_stream:write(OutputStream, Chunk)
-                    end, ok)
+                    ({Length, Chunk}, Total0) ->
+                        Total = Total0 + Length,
+                        validate_attachment_size(AttName, Total, MaxAttSize),
+                        couch_stream:write(OutputStream, Chunk),
+                        Total
+                    end, 0)
             end);
         AttLen ->
+            validate_attachment_size(AttName, AttLen, MaxAttSize),
             couch_db:with_stream(Fd, Att, fun(OutputStream) ->
                 write_streamed_attachment(OutputStream, Fun, AttLen)
             end)
@@ -680,6 +691,22 @@ upgrade_encoding(false) -> identity;
 upgrade_encoding(Encoding) -> Encoding.
 
 
+max_attachment_size() ->
+    case config:get("couchdb", "max_attachment_size", "infinity") of
+        "infinity" ->
+            infinity;
+        MaxAttSize ->
+            list_to_integer(MaxAttSize)
+    end.
+
+
+validate_attachment_size(AttName, AttSize, MaxAttSize)
+        when is_integer(AttSize),  AttSize > MaxAttSize ->
+    throw({request_entity_too_large, {attachment, AttName}});
+validate_attachment_size(_AttName, _AttSize, _MAxAttSize) ->
+    ok.
+
+
 -ifdef(TEST).
 -include_lib("eunit/include/eunit.hrl").
 
diff --git a/src/couch/src/couch_doc.erl b/src/couch/src/couch_doc.erl
index 381ad4b4f..2f792bf15 100644
--- a/src/couch/src/couch_doc.erl
+++ b/src/couch/src/couch_doc.erl
@@ -16,6 +16,7 @@
 -export([from_json_obj/1, from_json_obj_validate/1, to_json_obj/2,has_stubs/1, merge_stubs/2]).
 -export([validate_docid/1, get_validate_doc_fun/1]).
 -export([doc_from_multi_part_stream/2, doc_from_multi_part_stream/3]).
+-export([doc_from_multi_part_stream/4]).
 -export([doc_to_multi_part_stream/5, len_doc_to_multi_part_stream/4]).
 -export([restart_open_doc_revs/3]).
 -export([to_path/1]).
@@ -129,11 +130,24 @@ from_json_obj_validate(EJson) ->
     Doc = from_json_obj(EJson),
     case erlang:external_size(Doc#doc.body) =< MaxSize of
         true ->
+             validate_attachment_sizes(Doc#doc.atts),
              Doc;
         false ->
             throw({request_entity_too_large, Doc#doc.id})
     end.
 
+
+validate_attachment_sizes([]) ->
+    ok;
+validate_attachment_sizes(Atts) ->
+    MaxAttSize = couch_att:max_attachment_size(),
+    lists:foreach(fun(Att) ->
+         AttName = couch_att:fetch(name, Att),
+         AttSize = couch_att:fetch(att_len, Att),
+         couch_att:validate_attachment_size(AttName, AttSize, MaxAttSize)
+    end, Atts).
+
+
 from_json_obj({Props}) ->
     transfer_fields(Props, #doc{body=[]});
 
@@ -420,11 +434,19 @@ doc_from_multi_part_stream(ContentType, DataFun) ->
     doc_from_multi_part_stream(ContentType, DataFun, make_ref()).
 
 doc_from_multi_part_stream(ContentType, DataFun, Ref) ->
+    doc_from_multi_part_stream(ContentType, DataFun, Ref, true).
+
+doc_from_multi_part_stream(ContentType, DataFun, Ref, ValidateDocLimits) ->
     case couch_httpd_multipart:decode_multipart_stream(ContentType, DataFun, Ref) of
     {{started_open_doc_revs, NewRef}, Parser, _ParserRef} ->
         restart_open_doc_revs(Parser, Ref, NewRef);
     {{doc_bytes, Ref, DocBytes}, Parser, ParserRef} ->
-        Doc = from_json_obj_validate(?JSON_DECODE(DocBytes)),
+        Doc = case ValidateDocLimits of
+            true ->
+                from_json_obj_validate(?JSON_DECODE(DocBytes));
+            false ->
+                from_json_obj(?JSON_DECODE(DocBytes))
+        end,
         erlang:put(mochiweb_request_recv, true),
         % we'll send the Parser process ID to the remote nodes so they can
         % retrieve their own copies of the attachment data
diff --git a/src/couch/src/couch_httpd.erl b/src/couch/src/couch_httpd.erl
index faaf080d9..b3bbd5baa 100644
--- a/src/couch/src/couch_httpd.erl
+++ b/src/couch/src/couch_httpd.erl
@@ -885,6 +885,10 @@ error_info(file_exists) ->
         "created, the file already exists.">>};
 error_info(request_entity_too_large) ->
     {413, <<"too_large">>, <<"the request entity is too large">>};
+error_info({request_entity_too_large, {attachment, AttName}}) ->
+    {413, <<"attachment_too_large">>, AttName};
+error_info({request_entity_too_large, DocID}) ->
+    {413, <<"document_too_large">>, DocID};
 error_info(request_uri_too_long) ->
     {414, <<"too_long">>, <<"the request uri is too long">>};
 error_info({bad_ctype, Reason}) ->
diff --git a/src/couch/test/couch_doc_tests.erl b/src/couch/test/couch_doc_tests.erl
index d24cd67c0..5d0448a9e 100644
--- a/src/couch/test/couch_doc_tests.erl
+++ b/src/couch/test/couch_doc_tests.erl
@@ -131,6 +131,7 @@ mock_config_max_document_id_length() ->
     ok = meck:new(config, [passthrough]),
     meck:expect(config, get,
         fun("couchdb", "max_document_id_length", "infinity") -> "1024";
+           ("couchdb", "max_attachment_size", "infinity") -> "infinity";
             (Key, Val, Default) -> meck:passthrough([Key, Val, Default])
         end
     ).
diff --git a/src/couch/test/couchdb_attachments_tests.erl b/src/couch/test/couchdb_attachments_tests.erl
index 88f53a189..4536ba6b2 100644
--- a/src/couch/test/couchdb_attachments_tests.erl
+++ b/src/couch/test/couchdb_attachments_tests.erl
@@ -14,6 +14,7 @@
 
 -include_lib("couch/include/couch_eunit.hrl").
 -include_lib("couch/include/couch_db.hrl").
+-include_lib("mem3/include/mem3.hrl").
 
 -define(COMPRESSION_LEVEL, 8).
 -define(ATT_BIN_NAME, <<"logo.png">>).
@@ -515,6 +516,137 @@ should_create_compressible_att_with_ctype_params({Host, DbName}) ->
     end)}.
 
 
+compact_after_lowering_attachment_size_limit_test_() ->
+    {
+        "Compact after lowering attachment size limit",
+        {
+            foreach,
+            fun() ->
+                Ctx = test_util:start_couch(),
+                DbName = ?tempdb(),
+                {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
+                ok = couch_db:close(Db),
+                {Ctx, DbName}
+            end,
+            fun({Ctx, DbName}) ->
+                config:delete("couchdb", "max_attachment_size"),
+                ok = couch_server:delete(DbName, [?ADMIN_CTX]),
+                test_util:stop_couch(Ctx)
+            end,
+            [
+                fun should_compact_after_lowering_attachment_size_limit/1
+            ]
+        }
+    }.
+
+
+should_compact_after_lowering_attachment_size_limit({_Ctx, DbName}) ->
+    {timeout, ?TIMEOUT_EUNIT, ?_test(begin
+        {ok, Db1} = couch_db:open(DbName, [?ADMIN_CTX]),
+        Doc1 = #doc{id = <<"doc1">>, atts = att(1000)},
+        {ok, _} = couch_db:update_doc(Db1, Doc1, []),
+        couch_db:close(Db1),
+        config:set("couchdb", "max_attachment_size", "1", _Persist = false),
+        compact_db(DbName),
+        {ok, Db2} = couch_db:open_int(DbName, []),
+        {ok, Doc2} = couch_db:open_doc(Db2, <<"doc1">>),
+        couch_db:close(Db2),
+        [Att] = Doc2#doc.atts,
+        ?assertEqual(1000, couch_att:fetch(att_len, Att))
+    end)}.
+
+
+att(Size) when is_integer(Size), Size >= 1 ->
+    [couch_att:new([
+        {name, <<"att">>},
+        {type, <<"app/binary">>},
+        {att_len, Size},
+        {data, fun(_Bytes) ->
+            << <<"x">> || _ <- lists:seq(1, Size) >>
+        end}
+    ])].
+
+
+compact_db(DbName) ->
+    {ok, Db} = couch_db:open_int(DbName, []),
+    {ok, _CompactPid} = couch_db:start_compact(Db),
+    wait_compaction(DbName, "database", ?LINE),
+    ok = couch_db:close(Db).
+
+
+wait_compaction(DbName, Kind, Line) ->
+    WaitFun = fun() ->
+       case is_compaction_running(DbName) of
+           true -> wait;
+           false -> ok
+       end
+    end,
+    case test_util:wait(WaitFun, ?TIMEOUT) of
+        timeout ->
+            erlang:error({assertion_failed,
+                          [{module, ?MODULE},
+                           {line, Line},
+                           {reason, "Timeout waiting for "
+                                    ++ Kind
+                                    ++ " database compaction"}]});
+        _ ->
+            ok
+    end.
+
+
+is_compaction_running(DbName) ->
+    {ok, Db} = couch_db:open_int(DbName, []),
+    {ok, DbInfo} = couch_db:get_db_info(Db),
+    couch_db:close(Db),
+    couch_util:get_value(compact_running, DbInfo) =:= true.
+
+
+internal_replication_after_lowering_attachment_size_limit_test_() ->
+    {
+        "Internal replication after lowering max attachment size",
+        {
+            foreach,
+            fun() ->
+                Ctx = test_util:start_couch([mem3]),
+                SrcName = ?tempdb(),
+                {ok, SrcDb} = couch_db:create(SrcName, [?ADMIN_CTX]),
+                ok = couch_db:close(SrcDb),
+                TgtName = ?tempdb(),
+                {ok, TgtDb} = couch_db:create(TgtName, [?ADMIN_CTX]),
+                ok = couch_db:close(TgtDb),
+                {Ctx, SrcName, TgtName}
+            end,
+            fun({Ctx, SrcName, TgtName}) ->
+                config:delete("couchdb", "max_attachment_size"),
+                ok = couch_server:delete(SrcName, [?ADMIN_CTX]),
+                ok = couch_server:delete(TgtName, [?ADMIN_CTX]),
+                test_util:stop_couch(Ctx)
+            end,
+            [
+                fun should_replicate_after_lowering_attachment_size/1
+            ]
+        }
+    }.
+
+should_replicate_after_lowering_attachment_size({_Ctx, SrcName, TgtName}) ->
+    {timeout, ?TIMEOUT_EUNIT, ?_test(begin
+        {ok, SrcDb} = couch_db:open(SrcName, [?ADMIN_CTX]),
+        SrcDoc = #doc{id = <<"doc">>, atts = att(1000)},
+        {ok, _} = couch_db:update_doc(SrcDb, SrcDoc, []),
+        couch_db:close(SrcDb),
+        config:set("couchdb", "max_attachment_size", "1", _Persist = false),
+        % Create a pair of "fake" shards
+        SrcShard = #shard{name = SrcName, node = node()},
+        TgtShard = #shard{name = TgtName, node = node()},
+        mem3_rep:go(SrcShard, TgtShard, []),
+        {ok, TgtDb} = couch_db:open_int(TgtName, []),
+        {ok, TgtDoc} = couch_db:open_doc(TgtDb, <<"doc">>),
+        couch_db:close(TgtDb),
+        [Att] = TgtDoc#doc.atts,
+        ?assertEqual(1000, couch_att:fetch(att_len, Att))
+    end)}.
+
+
 get_json(Json, Path) ->
     couch_util:get_nested_json_value(Json, Path).
 
diff --git a/src/couch_replicator/src/couch_replicator_api_wrap.erl b/src/couch_replicator/src/couch_replicator_api_wrap.erl
index 91d7d7ae5..a2ef60fa3 100644
--- a/src/couch_replicator/src/couch_replicator_api_wrap.erl
+++ b/src/couch_replicator/src/couch_replicator_api_wrap.erl
@@ -711,10 +711,12 @@ receive_docs(Streamer, UserFun, Ref, UserAcc) ->
     {headers, Ref, Headers} ->
         case header_value("content-type", Headers) of
         {"multipart/related", _} = ContentType ->
+            % Skip document body and attachment size limits validation here
+            % since these should be validated by the replication target
             case couch_doc:doc_from_multi_part_stream(
                 ContentType,
                 fun() -> receive_doc_data(Streamer, Ref) end,
-                Ref) of
+                Ref, _ValidateDocLimits = false) of
             {ok, Doc, WaitFun, Parser} ->
                 case run_user_fun(UserFun, {ok, Doc}, UserAcc, Ref) of
                 {ok, UserAcc2} ->
diff --git a/src/couch_replicator/test/couch_replicator_attachments_too_large.erl b/src/couch_replicator/test/couch_replicator_attachments_too_large.erl
new file mode 100644
index 000000000..7fe84d2d9
--- /dev/null
+++ b/src/couch_replicator/test/couch_replicator_attachments_too_large.erl
@@ -0,0 +1,104 @@
+% 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_replicator_attachments_too_large).
+
+-include_lib("couch/include/couch_eunit.hrl").
+-include_lib("couch/include/couch_db.hrl").
+-include_lib("couch_replicator/src/couch_replicator.hrl").
+
+
+setup(_) ->
+    Ctx = test_util:start_couch([couch_replicator]),
+    Source = create_db(),
+    create_doc_with_attachment(Source, <<"doc">>, 1000),
+    Target = create_db(),
+    {Ctx, {Source, Target}}.
+
+
+teardown(_, {Ctx, {Source, Target}}) ->
+    delete_db(Source),
+    delete_db(Target),
+    config:delete("couchdb", "max_attachment_size"),
+    ok = test_util:stop_couch(Ctx).
+
+
+attachment_too_large_replication_test_() ->
+    Pairs = [{local, remote}, {remote, local}, {remote, remote}],
+    {
+        "Attachment size too large replication tests",
+        {
+            foreachx,
+            fun setup/1, fun teardown/2,
+            [{Pair, fun should_succeed/2} || Pair <- Pairs] ++
+            [{Pair, fun should_fail/2} || Pair <- Pairs]
+        }
+    }.
+
+
+should_succeed({From, To}, {_Ctx, {Source, Target}}) ->
+    RepObject = {[
+        {<<"source">>, db_url(From, Source)},
+        {<<"target">>, db_url(To, Target)}
+    ]},
+    config:set("couchdb", "max_attachment_size", "1000", _Persist = false),
+    {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+    ?_assertEqual(ok, couch_replicator_test_helper:compare_dbs(Source, Target)).
+
+
+should_fail({From, To}, {_Ctx, {Source, Target}}) ->
+    RepObject = {[
+        {<<"source">>, db_url(From, Source)},
+        {<<"target">>, db_url(To, Target)}
+    ]},
+    config:set("couchdb", "max_attachment_size", "999", _Persist = false),
+    {ok, _} = couch_replicator:replicate(RepObject, ?ADMIN_USER),
+    ?_assertError({badmatch, {not_found, missing}},
+        couch_replicator_test_helper:compare_dbs(Source, Target)).
+
+
+create_db() ->
+    DbName = ?tempdb(),
+    {ok, Db} = couch_db:create(DbName, [?ADMIN_CTX]),
+    ok = couch_db:close(Db),
+    DbName.
+
+
+create_doc_with_attachment(DbName, DocId, AttSize) ->
+    {ok, Db} = couch_db:open(DbName, [?ADMIN_CTX]),
+    Doc = #doc{id = DocId, atts = att(AttSize)},
+    {ok, _} = couch_db:update_doc(Db, Doc, []),
+    couch_db:close(Db),
+    ok.
+
+
+att(Size) when is_integer(Size), Size >= 1 ->
+    [couch_att:new([
+        {name, <<"att">>},
+        {type, <<"app/binary">>},
+        {att_len, Size},
+        {data, fun(_Bytes) ->
+            << <<"x">> || _ <- lists:seq(1, Size) >>
+        end}
+    ])].
+
+
+delete_db(DbName) ->
+    ok = couch_server:delete(DbName, [?ADMIN_CTX]).
+
+
+db_url(local, DbName) ->
+    DbName;
+db_url(remote, DbName) ->
+    Addr = config:get("httpd", "bind_address", "127.0.0.1"),
+    Port = mochiweb_socket_server:get(couch_httpd, port),
+    ?l2b(io_lib:format("http://~s:~b/~s", [Addr, Port, DbName])).
diff --git a/src/fabric/src/fabric_doc_attachments.erl b/src/fabric/src/fabric_doc_attachments.erl
index 8b8123fa9..47854d189 100644
--- a/src/fabric/src/fabric_doc_attachments.erl
+++ b/src/fabric/src/fabric_doc_attachments.erl
@@ -24,8 +24,8 @@ receiver(_Req, {unknown_transfer_encoding, Unknown}) ->
     exit({unknown_transfer_encoding, Unknown});
 receiver(Req, chunked) ->
     MiddleMan = spawn(fun() -> middleman(Req, chunked) end),
-    fun(4096, ChunkFun, ok) ->
-        write_chunks(MiddleMan, ChunkFun)
+    fun(4096, ChunkFun, State) ->
+        write_chunks(MiddleMan, ChunkFun, State)
     end;
 receiver(_Req, 0) ->
     <<"">>;
@@ -63,27 +63,29 @@ maybe_send_continue(#httpd{mochi_req = MochiReq} = Req) ->
         end
     end.
 
-write_chunks(MiddleMan, ChunkFun) ->
+write_chunks(MiddleMan, ChunkFun, State) ->
     MiddleMan ! {self(), gimme_data},
     Timeout = fabric_util:attachments_timeout(),
     receive
     {MiddleMan, ChunkRecordList} ->
         rexi:reply(attachment_chunk_received),
-        case flush_chunks(ChunkRecordList, ChunkFun) of
-        continue -> write_chunks(MiddleMan, ChunkFun);
-        done -> ok
+        case flush_chunks(ChunkRecordList, ChunkFun, State) of
+            {continue, NewState} ->
+                write_chunks(MiddleMan, ChunkFun, NewState);
+            {done, NewState} ->
+                NewState
         end
     after Timeout ->
         exit(timeout)
     end.
 
-flush_chunks([], _ChunkFun) ->
-    continue;
-flush_chunks([{0, _}], _ChunkFun) ->
-    done;
-flush_chunks([Chunk | Rest], ChunkFun) ->
-    ChunkFun(Chunk, ok),
-    flush_chunks(Rest, ChunkFun).
+flush_chunks([], _ChunkFun, State) ->
+    {continue, State};
+flush_chunks([{0, _}], _ChunkFun, State) ->
+    {done, State};
+flush_chunks([Chunk | Rest], ChunkFun, State) ->
+    NewState = ChunkFun(Chunk, State),
+    flush_chunks(Rest, ChunkFun, NewState).
 
 receive_unchunked_attachment(_Req, 0) ->
     ok;
diff --git a/src/fabric/src/fabric_doc_update.erl b/src/fabric/src/fabric_doc_update.erl
index 10e254ff5..214566d91 100644
--- a/src/fabric/src/fabric_doc_update.erl
+++ b/src/fabric/src/fabric_doc_update.erl
@@ -96,7 +96,9 @@ handle_message({not_found, no_db_file} = X, Worker, Acc0) ->
     Docs = couch_util:get_value(Worker, GroupedDocs),
     handle_message({ok, [X || _D <- Docs]}, Worker, Acc0);
 handle_message({bad_request, Msg}, _, _) ->
-    throw({bad_request, Msg}).
+    throw({bad_request, Msg});
+handle_message({request_entity_too_large, Entity}, _, _) ->
+    throw({request_entity_too_large, Entity}).
 
 before_doc_update(DbName, Docs, Opts) ->
     case {fabric_util:is_replicator_db(DbName), fabric_util:is_users_db(DbName)} of