diff --git a/src/couch/src/couch_db_split.erl b/src/couch/src/couch_db_split.erl
index d219e37..dce9187 100644
--- a/src/couch/src/couch_db_split.erl
+++ b/src/couch/src/couch_db_split.erl
@@ -62,7 +62,10 @@
             catch
                 throw:{target_create_error, DbName, Error, TargetDbs} ->
                     cleanup_targets(TargetDbs, Engine),
-                    {error, {target_create_error, DbName, Error}}
+                    {error, {target_create_error, DbName, Error}};
+                throw:{range_error, Context, DocId, TargetDbs} ->
+                    cleanup_targets(TargetDbs, Engine),
+                    {error, {range_error, Context, DocId, maps:keys(TargetDbs)}}
             after
                 couch_db:close(SourceDb)
             end;
@@ -208,10 +211,20 @@
     DelOpt = [{context, compaction}, sync],
     couch_db_engine:delete(Engine, RootDir, Filepath, DelOpt).
 
-pick_target(DocId, #state{} = State, #{} = Targets) ->
+pick_target(DocId, #state{} = State, #{} = Targets, Context) when
+    is_binary(DocId), is_atom(Context)
+->
     #state{pickfun = PickFun, hashfun = HashFun} = State,
-    Key = PickFun(DocId, maps:keys(Targets), HashFun),
-    {Key, maps:get(Key, Targets)}.
+    TargetRanges = maps:keys(Targets),
+    Key = PickFun(DocId, TargetRanges, HashFun),
+    case Key of
+        not_in_range ->
+            % We found a document, or purge info which doesn't hash to
+            % any of the target ranges. Stop, and raise a fatal exception.
+            throw({range_error, Context, DocId, Targets});
+        [B, E] when is_integer(B), is_integer(E), B =< E, is_map_key(Key, Targets) ->
+            {Key, map_get(Key, Targets)}
+    end.
 
 set_targets_update_seq(#state{targets = Targets} = State) ->
     Seq = couch_db:get_update_seq(State#state.source_db),
@@ -337,11 +350,23 @@
         false -> Target1
     end.
 
-purge_cb({_PSeq, _UUID, Id, _Revs} = PI, #state{targets = Targets} = State) ->
-    {Key, Target} = pick_target(Id, State, Targets),
-    MaxBuffer = State#state.max_buffer_size,
-    Target1 = acc_and_flush(PI, Target, MaxBuffer, fun commit_purge_infos/1),
-    {ok, State#state{targets = Targets#{Key => Target1}}}.
+purge_cb({PSeq, UUID, Id, _Revs} = PI, #state{} = State) ->
+    #state{source_db = Db, targets = Targets} = State,
+    try
+        {Key, Target} = pick_target(Id, State, Targets, purge_info_copy),
+        MaxBuffer = State#state.max_buffer_size,
+        Target1 = acc_and_flush(PI, Target, MaxBuffer, fun commit_purge_infos/1),
+        {ok, State#state{targets = Targets#{Key => Target1}}}
+    catch
+        throw:{range_error, purge_info_copy, _, _} ->
+            % Before 3.4, due to a bug in internal replicator, it was possible
+            % for purge infos which don't belong to the source shard to end up
+            % there during, or after shard splitting. We choose to emit a warning
+            % and ignore the misplaced purge info record.
+            LogMsg = "~p : ignore misplaced purge info pseq:~p uuid:~p doc_id:~p shard:~p",
+            couch_log:warning(LogMsg, [?MODULE, PSeq, UUID, Id, couch_db:name(Db)]),
+            {ok, State}
+    end.
 
 commit_purge_infos(#target{buffer = [], db = Db} = Target) ->
     Target#target{db = Db};
@@ -367,7 +392,7 @@
     changes_cb(FDI, State);
 changes_cb(#full_doc_info{id = Id} = FDI, #state{} = State) ->
     #state{source_db = SourceDb, targets = Targets} = State,
-    {Key, Target} = pick_target(Id, State, Targets),
+    {Key, Target} = pick_target(Id, State, Targets, changes),
     FDI1 = process_fdi(FDI, SourceDb, Target#target.db),
     MaxBuffer = State#state.max_buffer_size,
     Target1 = acc_and_flush(FDI1, Target, MaxBuffer, fun commit_docs/1),
@@ -500,7 +525,7 @@
                 <<?LOCAL_DOC_PREFIX, _/binary>> ->
                     % Users' and replicator app's checkpoints go to their
                     % respective shards based on the general hashing algorithm
-                    {Key, Target} = pick_target(Id, State, Acc),
+                    {Key, Target} = pick_target(Id, State, Acc, local_docs),
                     #target{buffer = Docs} = Target,
                     Acc#{Key => Target#target{buffer = [Doc | Docs]}}
             end,
diff --git a/src/mem3/src/mem3_rep.erl b/src/mem3/src/mem3_rep.erl
index d186f9e..3cc3814 100644
--- a/src/mem3/src/mem3_rep.erl
+++ b/src/mem3/src/mem3_rep.erl
@@ -314,12 +314,18 @@
             {ok, Count}
     end.
 
-pull_purges_multi(#acc{source = Source} = Acc0) ->
-    #acc{batch_size = Count, seq = UpdateSeq, targets = Targets0} = Acc0,
+pull_purges_multi(#acc{} = Acc0) ->
+    #acc{
+        source = Source,
+        targets = Targets0,
+        batch_size = Count,
+        seq = UpdateSeq,
+        hashfun = HashFun
+    } = Acc0,
     with_src_db(Acc0, fun(Db) ->
         Targets = maps:map(
             fun(_, #tgt{} = T) ->
-                pull_purges(Db, Count, Source, T)
+                pull_purges(Db, Count, Source, T, HashFun)
             end,
             reset_remaining(Targets0)
         ),
@@ -343,7 +349,7 @@
         end
     end).
 
-pull_purges(Db, Count, SrcShard, #tgt{} = Tgt0) ->
+pull_purges(Db, Count, #shard{} = SrcShard, #tgt{} = Tgt0, HashFun) ->
     #tgt{shard = TgtShard} = Tgt0,
     SrcUUID = couch_db:get_uuid(Db),
     #shard{node = TgtNode, name = TgtDbName} = TgtShard,
@@ -354,18 +360,33 @@
         Infos == [] ->
             ok;
         true ->
-            {ok, _} = couch_db:purge_docs(Db, Infos, [?REPLICATED_CHANGES]),
+            % When shard ranges are split it's possible to pull purges from a
+            % larger target range to a smaller source range, we don't want to
+            % pull purges which don't belong on the source, so we filter them
+            % out using the same pickfun which we use when picking documents
+            #shard{range = SrcRange} = SrcShard,
+            BelongsFun = fun({_UUID, Id, _Revs}) when is_binary(Id) ->
+                mem3_reshard_job:pickfun(Id, [SrcRange], HashFun) =:= SrcRange
+            end,
+            Infos1 = lists:filter(BelongsFun, Infos),
+            {ok, _} = couch_db:purge_docs(Db, Infos1, [?REPLICATED_CHANGES]),
             Body = purge_cp_body(SrcShard, TgtShard, ThroughSeq),
             mem3_rpc:save_purge_checkpoint(TgtNode, TgtDbName, LocalPurgeId, Body)
     end,
     Tgt#tgt{remaining = max(0, Remaining)}.
 
-push_purges_multi(#acc{source = SrcShard} = Acc) ->
-    #acc{batch_size = BatchSize, seq = UpdateSeq, targets = Targets0} = Acc,
+push_purges_multi(#acc{} = Acc) ->
+    #acc{
+        source = SrcShard,
+        targets = Targets0,
+        batch_size = BatchSize,
+        seq = UpdateSeq,
+        hashfun = HashFun
+    } = Acc,
     with_src_db(Acc, fun(Db) ->
         Targets = maps:map(
             fun(_, #tgt{} = T) ->
-                push_purges(Db, BatchSize, SrcShard, T)
+                push_purges(Db, BatchSize, SrcShard, T, HashFun)
             end,
             reset_remaining(Targets0)
         ),
@@ -385,9 +406,9 @@
         end
     end).
 
-push_purges(Db, BatchSize, SrcShard, Tgt) ->
+push_purges(Db, BatchSize, SrcShard, Tgt, HashFun) ->
     #tgt{shard = TgtShard, purgeid = LocalPurgeId} = Tgt,
-    #shard{node = TgtNode, name = TgtDbName} = TgtShard,
+    #shard{node = TgtNode, range = TgtRange, name = TgtDbName} = TgtShard,
     StartSeq =
         case couch_db:open_doc(Db, LocalPurgeId, []) of
             {ok, #doc{body = {Props}}} ->
@@ -396,15 +417,25 @@
                 Oldest = couch_db:get_oldest_purge_seq(Db),
                 erlang:max(0, Oldest - 1)
         end,
+    BelongsFun = fun(Id) when is_binary(Id) ->
+        mem3_reshard_job:pickfun(Id, [TgtRange], HashFun) =:= TgtRange
+    end,
     FoldFun = fun({PSeq, UUID, Id, Revs}, {Count, Infos, _}) ->
-        NewCount = Count + length(Revs),
-        NewInfos = [{UUID, Id, Revs} | Infos],
-        Status =
-            if
-                NewCount < BatchSize -> ok;
-                true -> stop
-            end,
-        {Status, {NewCount, NewInfos, PSeq}}
+        case BelongsFun(Id) of
+            true ->
+                NewCount = Count + length(Revs),
+                NewInfos = [{UUID, Id, Revs} | Infos],
+                Status =
+                    if
+                        NewCount < BatchSize -> ok;
+                        true -> stop
+                    end,
+                {Status, {NewCount, NewInfos, PSeq}};
+            false ->
+                % In case of split shard ranges, purges, like documents, will
+                % belong only to one target
+                {ok, {Count, Infos, PSeq}}
+        end
     end,
     InitAcc = {0, [], StartSeq},
     {ok, {_, Infos, ThroughSeq}} =
diff --git a/src/mem3/src/mem3_reshard_job.erl b/src/mem3/src/mem3_reshard_job.erl
index 25ff1cf..4c93bf9 100644
--- a/src/mem3/src/mem3_reshard_job.erl
+++ b/src/mem3/src/mem3_reshard_job.erl
@@ -265,6 +265,19 @@
     couch_log:error(Msg, [?MODULE, jobfmt(Job)]),
     kill_workers(Job),
     exit({error, missing_target});
+handle_worker_exit(#job{} = Job, _Pid, {error, {range_error, _, _, _}} = Reason) ->
+    Msg1 = "~p fatal range error job:~p error:~p",
+    couch_log:error(Msg1, [?MODULE, jobfmt(Job), Reason]),
+    kill_workers(Job),
+    case lists:member(Job#job.split_state, ?CLEAN_TARGET_STATES) of
+        true ->
+            Msg2 = "~p cleaning target after db was deleted ~p",
+            couch_log:error(Msg2, [?MODULE, jobfmt(Job)]),
+            reset_target(Job),
+            exit(Reason);
+        false ->
+            exit(Reason)
+    end;
 handle_worker_exit(#job{} = Job0, _Pid, Reason) ->
     couch_log:error("~p worker error ~p ~p", [?MODULE, jobfmt(Job0), Reason]),
     kill_workers(Job0),
diff --git a/src/mem3/test/eunit/mem3_reshard_test.erl b/src/mem3/test/eunit/mem3_reshard_test.erl
index be539b4..4376ee4 100644
--- a/src/mem3/test/eunit/mem3_reshard_test.erl
+++ b/src/mem3/test/eunit/mem3_reshard_test.erl
@@ -77,7 +77,9 @@
                     fun retries_work/1,
                     fun target_reset_in_initial_copy/1,
                     fun split_an_incomplete_shard_map/1,
-                    fun target_shards_are_locked/1
+                    fun target_shards_are_locked/1,
+                    fun doc_in_bad_range_on_source/1,
+                    fun purge_info_in_bad_range_on_source/1
                 ]
             }
         }
@@ -88,7 +90,7 @@
 split_one_shard(#{db1 := Db}) ->
     {timeout, ?TIMEOUT,
         ?_test(begin
-            DocSpec = #{docs => 10, delete => [5, 9], mrview => 1, local => 1},
+            DocSpec = #{docs => 10, delete => [5, 9], mrview => 1, local => 10},
             add_test_docs(Db, DocSpec),
 
             % Save documents before the split
@@ -107,6 +109,7 @@
 
             % Split the one shard
             [#shard{name = Shard}] = lists:sort(mem3:local_shards(Db)),
+            SrcDocIds = get_shard_doc_ids(Shard),
             {ok, JobId} = mem3_reshard:start_split_job(Shard),
             wait_state(JobId, completed),
 
@@ -117,6 +120,16 @@
             ?assertEqual([16#00000000, 16#7fffffff], R1),
             ?assertEqual([16#80000000, 16#ffffffff], R2),
 
+            % Check that docs are on the shards where they belong to
+            [#shard{name = SN1}, #shard{name = SN2}] = Shards1,
+            DocIds1 = get_shard_doc_ids(SN1),
+            DocIds2 = get_shard_doc_ids(SN2),
+            [?assert(mem3:belongs(SN1, Id)) || Id <- DocIds1],
+            [?assert(mem3:belongs(SN2, Id)) || Id <- DocIds2],
+
+            % None of the docs or purges were dropped
+            ?assertEqual(lists:sort(SrcDocIds), lists:sort(DocIds1 ++ DocIds2)),
+
             % Check metadata bits after the split
             ?assertEqual(942, get_revs_limit(Db)),
             ?assertEqual(943, get_purge_infos_limit(Db)),
@@ -173,6 +186,9 @@
 
             % Split the one shard
             [#shard{name = Shard}] = lists:sort(mem3:local_shards(Db)),
+            % Get purge infos before the split
+            SrcPurges = get_purge_infos(Shard),
+            SrcDocIds = get_shard_doc_ids(Shard),
             {ok, JobId} = mem3_reshard:start_split_job(Shard),
             wait_state(JobId, completed),
 
@@ -183,6 +199,21 @@
             ?assertEqual([16#00000000, 16#7fffffff], R1),
             ?assertEqual([16#80000000, 16#ffffffff], R2),
 
+            % Check that purges and docs are on the shards where they belong to
+            [#shard{name = SN1}, #shard{name = SN2}] = Shards1,
+            TgtPurges1 = get_purge_infos(SN1),
+            TgtPurges2 = get_purge_infos(SN2),
+            DocIds1 = get_shard_doc_ids(SN1),
+            DocIds2 = get_shard_doc_ids(SN2),
+            [?assert(mem3:belongs(SN1, Id)) || Id <- TgtPurges1],
+            [?assert(mem3:belongs(SN2, Id)) || Id <- TgtPurges2],
+            [?assert(mem3:belongs(SN1, Id)) || Id <- DocIds1],
+            [?assert(mem3:belongs(SN2, Id)) || Id <- DocIds2],
+
+            % None of the docs or purges were dropped
+            ?assertEqual(lists:sort(SrcDocIds), lists:sort(DocIds1 ++ DocIds2)),
+            ?assertEqual(lists:sort(SrcPurges), lists:sort(TgtPurges1 ++ TgtPurges2)),
+
             % Check metadata bits after the split
             ?assertEqual(10, get_purge_infos_limit(Db)),
 
@@ -619,6 +650,53 @@
             wait_state(JobId, completed)
         end)}.
 
+% Source somehow got a bad doc which doesn't belong there
+doc_in_bad_range_on_source(#{db1 := Db}) ->
+    {timeout, ?TIMEOUT,
+        ?_test(begin
+            DocSpec = #{docs => 10, mrview => 0, local => 1},
+            add_test_docs(Db, DocSpec),
+
+            % Split first shard
+            [#shard{name = Shard1}] = lists:sort(mem3:local_shards(Db)),
+            {ok, JobId1} = mem3_reshard:start_split_job(Shard1),
+            wait_state(JobId1, completed),
+
+            % Split the first range again but before doing that insert
+            % a doc in the shard with a doc id that wouldn't belong to
+            % any target ranges
+            [#shard{name = Shard2}, _] = lists:sort(mem3:local_shards(Db)),
+            add_shard_doc(Shard2, <<"4">>, [{<<"in_the_wrong">>, <<"shard_range">>}]),
+            {ok, JobId2} = mem3_reshard:start_split_job(Shard2),
+            wait_state(JobId2, failed),
+            {ok, {JobProps}} = mem3_reshard:job(JobId2),
+            StateInfo = proplists:get_value(state_info, JobProps),
+            ?assertMatch({[{reason, <<"{error,{range_error", _/binary>>}]}, StateInfo)
+        end)}.
+
+% Source has a bad doc but we expect that due to a bug in <3.4 so we
+% skip over it and move on
+purge_info_in_bad_range_on_source(#{db1 := Db}) ->
+    {timeout, ?TIMEOUT,
+        ?_test(begin
+            DocSpec = #{docs => 10, mrview => 0, local => 1},
+            add_test_docs(Db, DocSpec),
+
+            % Split first shard
+            [#shard{name = Shard1}] = lists:sort(mem3:local_shards(Db)),
+            {ok, JobId1} = mem3_reshard:start_split_job(Shard1),
+            wait_state(JobId1, completed),
+
+            % Split the first range again but before doing that insert
+            % a purge info in the shard with an id which wouldn't belong to
+            % any target ranges
+            [#shard{name = Shard2}, _] = lists:sort(mem3:local_shards(Db)),
+            PurgeInfo = {couch_uuids:new(), <<"4">>, [{1, <<"a">>}]},
+            add_shard_purge_info(Shard2, PurgeInfo),
+            {ok, JobId2} = mem3_reshard:start_split_job(Shard2),
+            wait_state(JobId2, completed)
+        end)}.
+
 intercept_state(State) ->
     TestPid = self(),
     meck:new(mem3_reshard_job, [passthrough]),
@@ -988,3 +1066,29 @@
             {data, Data}
         ])
     ].
+
+get_purge_infos(ShardName) when is_binary(ShardName) ->
+    FoldFun = fun({_Seq, _UUID, Id, _Revs}, Acc) -> {ok, [Id | Acc]} end,
+    couch_util:with_db(ShardName, fun(Db) ->
+        PSeq = max(0, couch_db:get_oldest_purge_seq(Db) - 1),
+        {ok, Res} = couch_db:fold_purge_infos(Db, PSeq, FoldFun, []),
+        Res
+    end).
+
+get_shard_doc_ids(ShardName) when is_binary(ShardName) ->
+    FoldFun = fun(#full_doc_info{id = Id}, Acc) -> {ok, [Id | Acc]} end,
+    couch_util:with_db(ShardName, fun(Db) ->
+        {ok, Res} = couch_db:fold_docs(Db, FoldFun, [], []),
+        Res
+    end).
+
+add_shard_doc(ShardName, DocId, Props) ->
+    couch_util:with_db(ShardName, fun(Db) ->
+        Doc = couch_doc:from_json_obj({[{<<"_id">>, DocId}] ++ Props}),
+        couch_db:update_doc(Db, Doc, [])
+    end).
+
+add_shard_purge_info(ShardName, {UUID, Id, Revs}) ->
+    couch_util:with_db(ShardName, fun(Db) ->
+        couch_db:purge_docs(Db, [{UUID, Id, Revs}])
+    end).