replicator always retry on attachment streaming failure

gistfile1.diff

diff --git a/src/couchdb/couch_api_wrap.erl b/src/couchdb/couch_api_wrap.erl
index aa481c7..63e607a 100644
--- a/src/couchdb/couch_api_wrap.erl
+++ b/src/couchdb/couch_api_wrap.erl
@@ -597,7 +597,7 @@ doc_from_multi_part_stream(ContentType, DataFun) ->
             started ->
                 unlink(Parser),
                 exit(Parser, kill),
-                restart_remote_open_doc_revs();
+                restart_att_streaming();
             {bytes, Bytes} ->
                 Bytes
             end
@@ -612,6 +612,15 @@ doc_from_multi_part_stream(ContentType, DataFun) ->
     end.
 
 
+restart_att_streaming() ->
+    receive
+    {bytes, _} ->
+        restart_att_streaming()
+    after 0 ->
+        throw({retry, att_streaming_failed})
+    end.
+
+
 changes_ev1(object_start, UserFun, UserAcc) ->
     fun(Ev) -> changes_ev2(Ev, UserFun, UserAcc) end.
 
diff --git a/src/couchdb/couch_replicator_doc_copier.erl b/src/couchdb/couch_replicator_doc_copier.erl
index 91b912c..90eaa4f 100644
--- a/src/couchdb/couch_replicator_doc_copier.erl
+++ b/src/couchdb/couch_replicator_doc_copier.erl
@@ -340,11 +340,16 @@ spawn_doc_reader(Source, Target, FetchParams) ->
     end).
 
 
-fetch_doc(Source, {Id, Revs, PAs, _Seq}, DocHandler, Acc) ->
+fetch_doc(Source, {Id, Revs, PAs, _Seq} = Params, DocHandler, Acc) ->
     try
         couch_api_wrap:open_doc_revs(
             Source, Id, Revs, [{atts_since, PAs}], DocHandler, Acc)
     catch
+    throw:{retry, att_streaming_failed} ->
+        ?LOG_ERROR("Retrying fetch and update of document `~p` due to failure "
+            "when streaming attachments. Missing revisions are: ~s",
+            [Id, couch_doc:revs_to_strs(Revs)]),
+        fetch_doc(Source, Params, DocHandler, Acc);
     throw:{missing_stub, _} ->
         ?LOG_ERROR("Retrying fetch and update of document `~p` due to out of "
             "sync attachment stubs. Missing revisions are: ~s",
@@ -515,6 +520,8 @@ flush_doc(Target, #doc{id = Id, revs = {Pos, [RevId | _]}} = Doc) ->
     catch
     throw:{missing_stub, _} = MissingStub ->
         throw(MissingStub);
+    throw:{retry, att_streaming_failed} = RetryAttFailed ->
+        throw(RetryAttFailed);
     throw:{Error, Reason} ->
         ?LOG_ERROR("Replicator: couldn't write document `~s`, revision `~s`,"
             " to target database `~s`. Error: `~s`, reason: `~s`.",