Implement attachment size limitsissue-769-1

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