SERVER-8598 allow CollectionManager to track pending chunks

Also includes fixes for MetadataLoader

11 files changed, 1159 insertions, 246 deletions

diff --git a/src/mongo/s/collection_metadata.cpp b/src/mongo/s/collection_metadata.cpp
index 0188d1a585e..d27c86c5ed6 100644
--- a/src/mongo/s/collection_metadata.cpp
+++ b/src/mongo/s/collection_metadata.cpp
@@ -17,7 +17,6 @@
 #include "mongo/s/collection_metadata.h"
 
 #include "mongo/bson/util/builder.h" // for StringBuilder
-#include "mongo/s/range_arithmetic.h"
 #include "mongo/util/mongoutils/str.h"
 
 namespace mongo {
@@ -28,9 +27,9 @@ namespace mongo {
 
     CollectionMetadata::~CollectionMetadata() { }
 
-    CollectionMetadata* CollectionMetadata::cloneMinus(const ChunkType& chunk,
-                                                     const ChunkVersion& newShardVersion,
-                                                     string* errMsg) const {
+    CollectionMetadata* CollectionMetadata::cloneMinusChunk( const ChunkType& chunk,
+                                                             const ChunkVersion& newShardVersion,
+                                                             string* errMsg ) const {
         // The error message string is optional.
         string dummy;
         if (errMsg == NULL) {
@@ -38,45 +37,68 @@ namespace mongo {
         }
 
         // Check that we have the exact chunk that will be subtracted.
-        if (!chunkExists(chunk, errMsg)) {
-            // Message was filled in by chunkExists().
+        if ( !rangeMapContains( _chunksMap, chunk.getMin(), chunk.getMax() ) ) {
+
+            *errMsg = stream() << "cannot remove chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << ", this shard does not contain the chunk";
+
+            if ( rangeMapOverlaps( _chunksMap, chunk.getMin(), chunk.getMax() ) ) {
+
+                RangeVector overlap;
+                getRangeMapOverlap( _chunksMap, chunk.getMin(), chunk.getMax(), &overlap );
+
+                *errMsg += stream() << " and it overlaps " << overlapToString( overlap );
+            }
+
+            warning() << *errMsg << endl;
             return NULL;
         }
 
         // If left with no chunks, check that the version is zero.
         if (_chunksMap.size() == 1) {
             if (newShardVersion.isSet()) {
-                *errMsg = stream() << "setting version to " << newShardVersion.toString()
-                                   << " on removing last chunk";
+
+                *errMsg = stream() << "cannot set shard version to non-zero value "
+                                   << newShardVersion.toString() << " when removing last chunk "
+                                   << rangeToString( chunk.getMin(), chunk.getMax() );
+
+                warning() << *errMsg << endl;
                 return NULL;
             }
         }
         // Can't move version backwards when subtracting chunks.  This is what guarantees that
         // no read or write would be taken once we subtract data from the current shard.
         else if (newShardVersion <= _shardVersion) {
-            *errMsg = stream() << "version " << newShardVersion.toString()
-                               << " not greater than " << _shardVersion.toString();
-            return NULL;
-        }
 
-        auto_ptr<CollectionMetadata> manager(new CollectionMetadata);
-        manager->_keyPattern = this->_keyPattern;
-        manager->_keyPattern.getOwned();
-        manager->_chunksMap = this->_chunksMap;
-        manager->_chunksMap.erase(chunk.getMin());
-        manager->_shardVersion = newShardVersion;
-        manager->_collVersion = newShardVersion > _collVersion ?
-                                   newShardVersion : this->_collVersion;
-        manager->fillRanges();
+            *errMsg = stream() << "cannot remove chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << " because the new shard version " << newShardVersion.toString()
+                               << " is not greater than the current shard version "
+                               << _shardVersion.toString();
 
-        dassert(manager->isValid());
+            warning() << *errMsg << endl;
+            return NULL;
+        }
 
-        return manager.release();
+        auto_ptr<CollectionMetadata> metadata( new CollectionMetadata );
+        metadata->_keyPattern = this->_keyPattern;
+        metadata->_keyPattern.getOwned();
+        metadata->_pendingMap = this->_pendingMap;
+        metadata->_chunksMap = this->_chunksMap;
+        metadata->_chunksMap.erase( chunk.getMin() );
+        metadata->_shardVersion = newShardVersion;
+        metadata->_collVersion =
+                newShardVersion > _collVersion ? newShardVersion : this->_collVersion;
+        metadata->fillRanges();
+
+        dassert(metadata->isValid());
+        return metadata.release();
     }
 
-    CollectionMetadata* CollectionMetadata::clonePlus(const ChunkType& chunk,
-                                                    const ChunkVersion& newShardVersion,
-                                                    string* errMsg) const {
+    CollectionMetadata* CollectionMetadata::clonePlusChunk( const ChunkType& chunk,
+                                                            const ChunkVersion& newShardVersion,
+                                                            string* errMsg ) const {
         // The error message string is optional.
         string dummy;
         if (errMsg == NULL) {
@@ -86,45 +108,147 @@ namespace mongo {
         // It is acceptable to move version backwards (e.g., undoing a migration that went bad
         // during commit) but only cloning away the last chunk may reset the version to 0.
         if (!newShardVersion.isSet()) {
-            *errMsg =  "version can't be set to zero";
+
+            *errMsg = stream() << "cannot add chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << " with zero shard version";
+
+            warning() << *errMsg << endl;
             return NULL;
         }
 
         // Check that there isn't any chunk on the interval to be added.
-       if (!_chunksMap.empty()) {
-           RangeMap::const_iterator it = _chunksMap.lower_bound(chunk.getMax());
-            if (it != _chunksMap.begin()) {
-                --it;
-            }
-            if (rangeOverlaps(chunk.getMin(), chunk.getMax(), it->first, it->second)) {
-                *errMsg = stream() << "ranges overlap, "
-                                   << "requested: " << chunk.getMin()
-                                   <<" -> " << chunk.getMax() << " "
-                                   << "existing: " << it->first.toString()
-                                   << " -> " + it->second.toString();
-                return NULL;
+        if ( rangeMapOverlaps( _chunksMap, chunk.getMin(), chunk.getMax() ) ) {
+
+            RangeVector overlap;
+            getRangeMapOverlap( _chunksMap, chunk.getMin(), chunk.getMax(), &overlap );
+
+            *errMsg = stream() << "cannot add chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << " because the chunk overlaps " << overlapToString( overlap );
+
+            warning() << *errMsg << endl;
+            return NULL;
+        }
+
+        auto_ptr<CollectionMetadata> metadata( new CollectionMetadata );
+        metadata->_keyPattern = this->_keyPattern;
+        metadata->_keyPattern.getOwned();
+        metadata->_pendingMap = this->_pendingMap;
+        metadata->_chunksMap = this->_chunksMap;
+        metadata->_chunksMap.insert( make_pair( chunk.getMin().getOwned(),
+                                                chunk.getMax().getOwned() ) );
+        metadata->_shardVersion = newShardVersion;
+        metadata->_collVersion =
+                newShardVersion > _collVersion ? newShardVersion : this->_collVersion;
+        metadata->fillRanges();
+
+        dassert(metadata->isValid());
+        return metadata.release();
+    }
+
+    CollectionMetadata* CollectionMetadata::cloneMinusPending( const ChunkType& pending,
+                                                               string* errMsg ) const {
+        // The error message string is optional.
+        string dummy;
+        if ( errMsg == NULL ) {
+            errMsg = &dummy;
+        }
+
+        // Check that we have the exact chunk that will be subtracted.
+        if ( !rangeMapContains( _pendingMap, pending.getMin(), pending.getMax() ) ) {
+
+            *errMsg = stream() << "cannot remove pending chunk "
+                               << rangeToString( pending.getMin(), pending.getMax() )
+                               << ", this shard does not contain the chunk";
+
+            if ( rangeMapOverlaps( _pendingMap, pending.getMin(), pending.getMax() ) ) {
+
+                RangeVector overlap;
+                getRangeMapOverlap( _pendingMap, pending.getMin(), pending.getMax(), &overlap );
+
+                *errMsg += stream() << " and it overlaps " << overlapToString( overlap );
             }
+
+            warning() << *errMsg << endl;
+            return NULL;
+        }
+
+        auto_ptr<CollectionMetadata> metadata( new CollectionMetadata );
+        metadata->_keyPattern = this->_keyPattern;
+        metadata->_keyPattern.getOwned();
+        metadata->_pendingMap = this->_pendingMap;
+        metadata->_pendingMap.erase( pending.getMin() );
+        metadata->_chunksMap = this->_chunksMap;
+        metadata->_rangesMap = this->_rangesMap;
+        metadata->_shardVersion = _shardVersion;
+        metadata->_collVersion = _collVersion;
+
+        dassert(metadata->isValid());
+        return metadata.release();
+    }
+
+    CollectionMetadata* CollectionMetadata::clonePlusPending( const ChunkType& pending,
+                                                              string* errMsg ) const {
+        // The error message string is optional.
+        string dummy;
+        if ( errMsg == NULL ) {
+            errMsg = &dummy;
+        }
+
+        if ( rangeMapOverlaps( _chunksMap, pending.getMin(), pending.getMax() ) ) {
+
+            RangeVector overlap;
+            getRangeMapOverlap( _chunksMap, pending.getMin(), pending.getMax(), &overlap );
+
+            *errMsg = stream() << "cannot add pending chunk "
+                               << rangeToString( pending.getMin(), pending.getMax() )
+                               << " because the chunk overlaps " << overlapToString( overlap );
+
+            warning() << *errMsg << endl;
+            return NULL;
         }
 
-        auto_ptr<CollectionMetadata> manager(new CollectionMetadata);
-        manager->_keyPattern = this->_keyPattern;
-        manager->_keyPattern.getOwned();
-        manager->_chunksMap = this->_chunksMap;
-        manager->_chunksMap.insert(make_pair(chunk.getMin().getOwned(), chunk.getMax().getOwned()));
-        manager->_shardVersion = newShardVersion;
-        manager->_collVersion = newShardVersion > _collVersion ?
-                                   newShardVersion : this->_collVersion;
-        manager->fillRanges();
+        auto_ptr<CollectionMetadata> metadata( new CollectionMetadata );
+        metadata->_keyPattern = this->_keyPattern;
+        metadata->_keyPattern.getOwned();
+        metadata->_pendingMap = this->_pendingMap;
+        metadata->_chunksMap = this->_chunksMap;
+        metadata->_rangesMap = this->_rangesMap;
+        metadata->_shardVersion = _shardVersion;
+        metadata->_collVersion = _collVersion;
+
+        // If there are any pending chunks on the interval to be added this is ok, since pending
+        // chunks aren't officially tracked yet and something may have changed on servers we do not
+        // see yet.
+        // We remove any chunks we overlap, the remote request starting a chunk migration must have
+        // been authoritative.
+
+        if ( rangeMapOverlaps( _pendingMap, pending.getMin(), pending.getMax() ) ) {
+
+            RangeVector pendingOverlap;
+            getRangeMapOverlap( _pendingMap, pending.getMin(), pending.getMax(), &pendingOverlap );
+
+            warning() << "new pending chunk " << rangeToString( pending.getMin(), pending.getMax() )
+                      << " overlaps existing pending chunks " << overlapToString( pendingOverlap )
+                      << ", a migration may not have completed" << endl;
+
+            for ( RangeVector::iterator it = pendingOverlap.begin(); it != pendingOverlap.end();
+                    ++it ) {
+                metadata->_pendingMap.erase( it->first );
+            }
+        }
 
-        dassert(manager->isValid());
+        metadata->_pendingMap.insert( make_pair( pending.getMin(), pending.getMax() ) );
 
-        return manager.release();
+        dassert(metadata->isValid());
+        return metadata.release();
     }
 
-    CollectionMetadata* CollectionMetadata::cloneSplit(const ChunkType& chunk,
-                                                     const vector<BSONObj>& splitKeys,
-                                                     const ChunkVersion& newShardVersion,
-                                                     string* errMsg) const {
+    CollectionMetadata* CollectionMetadata::cloneSplit( const ChunkType& chunk,
+                                                        const vector<BSONObj>& splitKeys,
+                                                        const ChunkVersion& newShardVersion,
+                                                        string* errMsg ) const {
         // The error message string is optional.
         string dummy;
         if (errMsg == NULL) {
@@ -140,50 +264,74 @@ namespace mongo {
         // TODO drop the uniqueness constraint and tighten the check below so that only the
         // minor portion of version changes
         if (newShardVersion <= _shardVersion) {
-            *errMsg = stream()<< "version " << newShardVersion.toString()
-                              << " not greater than " << _shardVersion.toString();
+
+            *errMsg = stream() << "cannot split chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << ", new shard version "
+                               << newShardVersion.toString()
+                               << " is not greater than current version "
+                               << _shardVersion.toString();
+
+            warning() << *errMsg << endl;
             return NULL;
         }
 
-        // Check that we have the exact chunk that will be split and that the split point is
-        // valid.
-        if (!chunkExists(chunk, errMsg)) {
+        // Check that we have the exact chunk that will be subtracted.
+        if ( !rangeMapContains( _chunksMap, chunk.getMin(), chunk.getMax() ) ) {
+
+            *errMsg = stream() << "cannot split chunk "
+                               << rangeToString( chunk.getMin(), chunk.getMax() )
+                               << ", this shard does not contain the chunk";
+
+            if ( rangeMapOverlaps( _chunksMap, chunk.getMin(), chunk.getMax() ) ) {
+
+                RangeVector overlap;
+                getRangeMapOverlap( _chunksMap, chunk.getMin(), chunk.getMax(), &overlap );
+
+                *errMsg += stream() << " and it overlaps " << overlapToString( overlap );
+            }
+
+            warning() << *errMsg << endl;
             return NULL;
         }
 
-        for (vector<BSONObj>::const_iterator it = splitKeys.begin();
-             it != splitKeys.end();
-             ++it) {
+        // Check that the split key is valid
+        for ( vector<BSONObj>::const_iterator it = splitKeys.begin(); it != splitKeys.end(); ++it )
+        {
             if (!rangeContains(chunk.getMin(), chunk.getMax(), *it)) {
-                *errMsg = stream() << "can split " << chunk.getMin()
-                                   << " -> " << chunk.getMax() << " on " << *it;
+
+                *errMsg = stream() << "cannot split chunk "
+                                   << rangeToString( chunk.getMin(), chunk.getMax() ) << " at key "
+                                   << *it;
+
+                warning() << *errMsg << endl;
                 return NULL;
             }
         }
 
-        auto_ptr<CollectionMetadata> manager(new CollectionMetadata);
-        manager->_keyPattern = this->_keyPattern;
-        manager->_keyPattern.getOwned();
-        manager->_chunksMap = this->_chunksMap;
-        manager->_shardVersion = newShardVersion; // will increment 2nd, 3rd,... chunks below
+        auto_ptr<CollectionMetadata> metadata(new CollectionMetadata);
+        metadata->_keyPattern = this->_keyPattern;
+        metadata->_keyPattern.getOwned();
+        metadata->_pendingMap = this->_pendingMap;
+        metadata->_chunksMap = this->_chunksMap;
+        metadata->_shardVersion = newShardVersion; // will increment 2nd, 3rd,... chunks below
 
         BSONObj startKey = chunk.getMin();
-        for (vector<BSONObj>::const_iterator it = splitKeys.begin();
-             it != splitKeys.end();
-             ++it) {
+        for ( vector<BSONObj>::const_iterator it = splitKeys.begin(); it != splitKeys.end();
+                ++it ) {
             BSONObj split = *it;
-            manager->_chunksMap[chunk.getMin()] = split.getOwned();
-            manager->_chunksMap.insert(make_pair(split.getOwned(), chunk.getMax().getOwned()));
-            manager->_shardVersion.incMinor();
+            metadata->_chunksMap[chunk.getMin()] = split.getOwned();
+            metadata->_chunksMap.insert( make_pair( split.getOwned(), chunk.getMax().getOwned() ) );
+            metadata->_shardVersion.incMinor();
             startKey = split;
         }
 
-        manager->_collVersion = manager->_shardVersion > _collVersion ?
-                        manager->_shardVersion : this->_collVersion;
-        manager->fillRanges();
+        metadata->_collVersion =
+                metadata->_shardVersion > _collVersion ? metadata->_shardVersion : _collVersion;
+        metadata->fillRanges();
 
-        dassert(manager->isValid());
-        return manager.release();
+        dassert(metadata->isValid());
+        return metadata.release();
     }
 
     bool CollectionMetadata::keyBelongsToMe( const BSONObj& key ) const {
@@ -202,8 +350,9 @@ namespace mongo {
 
         bool good = rangeContains( it->first, it->second, key );
 
+#ifdef _DEBUG
         // Logs if in debugging mode and the point doesn't belong here.
-        if ( dcompare(!good) ) {
+        if ( !good ) {
             log() << "bad: " << key << " " << it->first << " " << key.woCompare( it->first ) << " "
                   << key.woCompare( it->second ) << endl;
 
@@ -211,10 +360,28 @@ namespace mongo {
                 log() << "\t" << i->first << "\t" << i->second << "\t" << endl;
             }
         }
+#endif
 
         return good;
     }
 
+    bool CollectionMetadata::keyIsPending( const BSONObj& key ) const {
+        // If we aren't sharded, then the key is never pending (though it belongs-to-me)
+        if ( _keyPattern.isEmpty() ) {
+            return false;
+        }
+
+        if ( _pendingMap.size() <= 0 ) {
+            return false;
+        }
+
+        RangeMap::const_iterator it = _pendingMap.upper_bound( key );
+        if ( it != _pendingMap.begin() ) it--;
+
+        bool isPending = rangeContains( it->first, it->second, key );
+        return isPending;
+    }
+
     bool CollectionMetadata::getNextChunk(const BSONObj& lookupKey,
                                          ChunkType* chunk) const {
         if (_chunksMap.empty()) {
@@ -265,29 +432,6 @@ namespace mongo {
         return true;
     }
 
-    bool CollectionMetadata::chunkExists(const ChunkType& chunk,
-                                        string* errMsg) const {
-        RangeMap::const_iterator it = _chunksMap.find(chunk.getMin());
-        if (it == _chunksMap.end()) {
-            *errMsg =  stream() << "couldn't find chunk " << chunk.getMin()
-                                << "->" << chunk.getMax();
-            return false;
-        }
-
-        if (it->second.woCompare(chunk.getMax()) != 0) {
-            *errMsg = stream() << "ranges differ, "
-                               << "requested: "  << chunk.getMin()
-                               << " -> " << chunk.getMax() << " "
-                               << "existing: "
-                               << ((it == _chunksMap.end()) ?
-                                   "<empty>" :
-                                   it->first.toString() + " -> " + it->second.toString());
-            return false;
-        }
-
-        return true;
-    }
-
     void CollectionMetadata::fillRanges() {
         if (_chunksMap.empty())
             return;
@@ -322,4 +466,20 @@ namespace mongo {
         _rangesMap.insert(make_pair(min, max));
     }
 
+    string CollectionMetadata::rangeToString( const BSONObj& inclusiveLower,
+                                              const BSONObj& exclusiveUpper ) const {
+        stringstream ss;
+        ss << "[" << inclusiveLower.toString() << ", " << exclusiveUpper.toString() << ")";
+        return ss.str();
+    }
+
+    string CollectionMetadata::overlapToString( RangeVector overlap ) const {
+        stringstream ss;
+        for ( RangeVector::const_iterator it = overlap.begin(); it != overlap.end(); ++it ) {
+            if ( it != overlap.begin() ) ss << ", ";
+            ss << rangeToString( it->first, it->second );
+        }
+        return ss.str();
+    }
+
 } // namespace mongo
diff --git a/src/mongo/s/collection_metadata.h b/src/mongo/s/collection_metadata.h
index b87c980a2e0..fe79daa45d5 100644
--- a/src/mongo/s/collection_metadata.h
+++ b/src/mongo/s/collection_metadata.h
@@ -19,6 +19,7 @@
 #include "mongo/base/disallow_copying.h"
 #include "mongo/db/jsobj.h"
 #include "mongo/s/chunk_version.h"
+#include "mongo/s/range_arithmetic.h"
 #include "mongo/s/type_chunk.h"
 
 namespace mongo {
@@ -30,7 +31,7 @@ namespace mongo {
     typedef shared_ptr<const CollectionMetadata> CollectionMetadataPtr;
 
     /**
-     * The collection manager has metadata information about a collection, in particular the
+     * The collection metadata has metadata information about a collection, in particular the
      * sharding information. It's main goal in life is to be capable of answering if a certain
      * document belongs to it or not. (In some scenarios such as chunk migration, a given
      * document is in a shard but cannot be accessed.)
@@ -52,35 +53,57 @@ namespace mongo {
         //
 
         /**
-         * Returns a new manager's instance based on 'this's state by removing 'chunk'.
+         * Returns a new metadata's instance based on 'this's state by removing 'chunk'.
          * When cloning away the last chunk, 'newShardVersion' must be zero. In any case,
-         * the caller owns the new manager when the cloning is successful.
+         * the caller owns the new metadata when the cloning is successful.
          *
-         * If a new manager can't be created, returns NULL and fills in 'errMsg', if it was
+         * If a new metadata can't be created, returns NULL and fills in 'errMsg', if it was
          * provided.
          */
-        CollectionMetadata* cloneMinus( const ChunkType& chunk,
-                                        const ChunkVersion& newShardVersion,
-                                        string* errMsg ) const;
+        CollectionMetadata* cloneMinusChunk( const ChunkType& chunk,
+                                             const ChunkVersion& newShardVersion,
+                                             string* errMsg ) const;
+
+        /**
+         * Returns a new metadata's instance based on 'this's state by adding 'chunk'. The new
+         * metadata can never be zero, though (see cloneMinus). The caller owns the new metadata.
+         *
+         * If a new metadata can't be created, returns NULL and fills in 'errMsg', if it was
+         * provided.
+         */
+        CollectionMetadata* clonePlusChunk( const ChunkType& chunk,
+                                            const ChunkVersion& newShardVersion,
+                                            string* errMsg ) const;
 
         /**
-         * Returns a new manager's instance based on 'this's state by adding 'chunk'. The new
-         * manager can never be zero, though (see cloneMinus). The caller owns the new manager.
+         * Returns a new metadata's instance based on 'this's state by removing a 'pending' chunk.
+         *
+         * The shard and collection version of the new metadata are unaffected.  The caller owns the
+         * new metadata.
          *
-         * If a new manager can't be created, returns NULL and fills in 'errMsg', if it was
+         * If a new metadata can't be created, returns NULL and fills in 'errMsg', if it was
          * provided.
          */
-        CollectionMetadata* clonePlus( const ChunkType& chunk,
-                                       const ChunkVersion& newShardVersion,
-                                       string* errMsg ) const;
+        CollectionMetadata* cloneMinusPending( const ChunkType& pending, string* errMsg ) const;
 
         /**
-         * Returns a new manager's instance by splitting an existing 'chunk' at the points
+         * Returns a new metadata's instance based on 'this's state by adding a 'pending' chunk.
+         *
+         * The shard and collection version of the new metadata are unaffected.  The caller owns the
+         * new metadata.
+         *
+         * If a new metadata can't be created, returns NULL and fills in 'errMsg', if it was
+         * provided.
+         */
+        CollectionMetadata* clonePlusPending( const ChunkType& pending, string* errMsg ) const;
+
+        /**
+         * Returns a new metadata's instance by splitting an existing 'chunk' at the points
          * describe by 'splitKeys'. The first resulting chunk will have 'newShardVersion' and
          * subsequent one would have that with the minor version incremented at each chunk. The
-         * caller owns the manager.
+         * caller owns the metadata.
          *
-         * If a new manager can't be created, returns NULL and fills in 'errMsg', if it was
+         * If a new metadata can't be created, returns NULL and fills in 'errMsg', if it was
          * provided.
          */
         CollectionMetadata* cloneSplit( const ChunkType& chunk,
@@ -93,19 +116,25 @@ namespace mongo {
         //
 
         /**
-         * Returns true the document key 'key' belongs to this chunkset. Recall that documents of
+         * Returns true if the document key 'key' belongs to this chunkset. Recall that documents of
          * an in-flight chunk migration may be present and should not be considered part of the
          * collection / chunkset yet. Key must be the full shard key.
          */
         bool keyBelongsToMe( const BSONObj& key ) const;
 
         /**
+         * Returns true if the document key 'key' is or has been migrated to this shard, and may
+         * belong to us after a subsequent config reload.  Key must be the full shard key.
+         */
+        bool keyIsPending( const BSONObj& key ) const;
+
+        /**
          * Given the chunk's min key (or empty doc) in 'lookupKey', gets the boundaries of the
          * chunk following that one (the first), and fills in 'foundChunk' with those
          * boundaries.  If the next chunk happens to be the last one, returns true otherwise
          * false.
          *
-         * @param lookupKey passing a doc that does not belong to this manager is undefined.
+         * @param lookupKey passing a doc that does not belong to this metadata is undefined.
          *     An empty doc is special and the chunk with the lowest range will be set on
          *     foundChunk.
          */
@@ -131,11 +160,15 @@ namespace mongo {
             return _chunksMap.size();
         }
 
+        std::size_t getNumPending() const {
+            return _pendingMap.size();
+        }
+
         string toString() const;
 
         /**
          * Use the MetadataLoader to fill the empty metadata from the config server, or use
-         * clone*() methods to use existing managers to build new ones.
+         * clone*() methods to use existing metadatas to build new ones.
          *
          * Unless you are the MetadataLoader or a test you should probably not be using this
          * directly.
@@ -155,14 +188,21 @@ namespace mongo {
         // sharded state below, for when the collection gets sharded
         //
 
-        // highest ChunkVersion for which this manager's information is accurate
+        // highest ChunkVersion for which this metadata's information is accurate
         ChunkVersion _shardVersion;
 
         // key pattern for chunks under this range
         BSONObj _keyPattern;
 
-        // a map from a min key into the chunk's (or range's) max boundary
-        typedef map<BSONObj, BSONObj, BSONObjCmp> RangeMap;
+        //
+        // RangeMaps represent chunks by mapping the min key to the chunk's max key, allowing
+        // efficient lookup and intersection.
+        //
+
+        // Map of ranges of chunks that are migrating but have not been confirmed added yet
+        RangeMap _pendingMap;
+
+        // Map of chunks tracked by this shard
         RangeMap _chunksMap;
 
         // A second map from a min key into a range or contiguous chunks. The map is redundant
@@ -171,19 +211,25 @@ namespace mongo {
         RangeMap _rangesMap;
 
         /**
-         * Returns true if this manager was loaded with all necessary information.
+         * Returns true if this metadata was loaded with all necessary information.
          */
         bool isValid() const;
 
         /**
-         * Returns true if 'chunk' exist in this * collections's chunkset.
+         * Try to find chunks that are adjacent and record these intervals in the _rangesMap
          */
-        bool chunkExists( const ChunkType& chunk, string* errMsg ) const;
+        void fillRanges();
 
         /**
-         * Try to find chunks that are adjacent and record these intervals in the _rangesMap
+         * String representation of [inclusiveLower, exclusiveUpper)
          */
-        void fillRanges();
+        std::string rangeToString( const BSONObj& inclusiveLower,
+                                   const BSONObj& exclusiveUpper ) const;
+
+        /**
+         * String representation of overlapping ranges as a list "[range1),[range2),..."
+         */
+        std::string overlapToString( RangeVector overlap ) const;
 
     };
 
diff --git a/src/mongo/s/collection_metadata_test.cpp b/src/mongo/s/collection_metadata_test.cpp
index 839feb0843b..ac502e0ed1d 100644
--- a/src/mongo/s/collection_metadata_test.cpp
+++ b/src/mongo/s/collection_metadata_test.cpp
@@ -61,11 +61,30 @@ namespace {
             MockConnRegistry::get()->addServer( _dummyConfig.get() );
 
             OID epoch = OID::gen();
-            _dummyConfig->insert( CollectionType::ConfigNS, BSON(CollectionType::ns("test.foo") <<
-                    CollectionType::keyPattern(BSON("a" << 1)) <<
-                    CollectionType::unique(false) <<
-                    CollectionType::updatedAt(1ULL) <<
-                    CollectionType::epoch(epoch)) );
+
+            CollectionType collType;
+            collType.setNS( "test.foo" );
+            collType.setKeyPattern( BSON("a" << 1) );
+            collType.setUnique( false );
+            collType.setUpdatedAt( 1ULL );
+            collType.setEpoch( epoch );
+            string errMsg;
+            ASSERT( collType.isValid( &errMsg ) );
+
+            _dummyConfig->insert( CollectionType::ConfigNS, collType.toBSON() );
+
+            // Need a chunk on another shard, otherwise the chunks are invalid in general and we
+            // can't load metadata
+            ChunkType chunkType;
+            chunkType.setNS( "test.foo");
+            chunkType.setShard( "shard0001" );
+            chunkType.setMin( BSON( "a" << MINKEY ) );
+            chunkType.setMax( BSON( "a" << MAXKEY ) );
+            chunkType.setVersion( ChunkVersion( 1, 0, epoch ) );
+            chunkType.setName( OID::gen().toString() );
+            ASSERT( chunkType.isValid( &errMsg ) );
+
+            _dummyConfig->insert( ChunkType::ConfigNS, chunkType.toBSON() );
 
             ConnectionString configLoc( CONFIG_HOST_PORT );
             MetadataLoader loader( configLoc );
@@ -75,6 +94,7 @@ namespace {
                                                            NULL,
                                                            &_metadata );
             ASSERT( status.isOK() );
+            ASSERT_EQUALS( 0u, _metadata.getNumChunks() );
         }
 
         void tearDown() {
@@ -95,7 +115,7 @@ namespace {
         ASSERT_FALSE( getCollMetadata().keyBelongsToMe(BSON("a" << 10)) );
     }
 
-    TEST_F(NoChunkFixture, CompoudKeyBelongsToMe) {
+    TEST_F(NoChunkFixture, CompoundKeyBelongsToMe) {
         ASSERT_FALSE( getCollMetadata().keyBelongsToMe(BSON("a" << 1 << "b" << 2)) );
     }
 
@@ -111,9 +131,9 @@ namespace {
 
         string errMsg;
         const ChunkVersion version( 99, 0, OID() );
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlus( chunk,
-                                                                            version,
-                                                                            &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlusChunk( chunk,
+                                                                                 version,
+                                                                                 &errMsg ) );
 
         ASSERT( errMsg.empty() );
         ASSERT_EQUALS( 1u, cloned->getNumChunks() );
@@ -128,16 +148,224 @@ namespace {
         chunk.setMax( BSON("a" << 20) );
 
         string errMsg;
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlus( chunk,
-                                                                            ChunkVersion( 0,
-                                                                                          0,
-                                                                                          OID() ),
-                                                                            &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata() // br
+                .clonePlusChunk( chunk, ChunkVersion( 0, 0, OID() ), &errMsg ) );
 
         ASSERT( cloned == NULL );
         ASSERT_FALSE( errMsg.empty() );
     }
 
+    TEST_F(NoChunkFixture, NoPendingChunks) {
+        ASSERT( !getCollMetadata().keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !getCollMetadata().keyIsPending(BSON( "a" << 25 )) );
+    }
+
+    TEST_F(NoChunkFixture, FirstPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 10 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 20 )) );
+    }
+
+    TEST_F(NoChunkFixture, EmptyMultiPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        chunk.setMin( BSON("a" << 40) );
+        chunk.setMax( BSON("a" << 50) );
+
+        cloned.reset( cloned->clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 45 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 55 )) );
+    }
+
+    TEST_F(NoChunkFixture, MinusPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        cloned.reset( cloned->cloneMinusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 25 )) );
+    }
+
+    TEST_F(NoChunkFixture, OverlappingPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 40) );
+
+        cloned.reset( cloned->clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 35 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 45 )) );
+    }
+
+    TEST_F(NoChunkFixture, OverlappingPendingChunks) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        chunk.setMin( BSON("a" << 30) );
+        chunk.setMax( BSON("a" << 50) );
+
+        cloned.reset( cloned->clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 40) );
+
+        cloned.reset( cloned->clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( cloned->keyIsPending(BSON( "a" << 35 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 45 )) );
+    }
+
+    TEST_F(NoChunkFixture, MinusInvalidPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().cloneMinusPending( chunk, &errMsg ) );
+
+        ASSERT_NOT_EQUALS( errMsg, "" );
+        ASSERT( cloned == NULL );
+    }
+
+    TEST_F(NoChunkFixture, MinusOverlappingPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        chunk.setMin( BSON("a" << 15) );
+        chunk.setMax( BSON("a" << 35) );
+
+        cloned.reset( cloned->cloneMinusPending( chunk, &errMsg ) );
+
+        ASSERT_NOT_EQUALS( errMsg, "" );
+        ASSERT( cloned == NULL );
+    }
+
+    TEST_F(NoChunkFixture, PlusChunkWithPending) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 25 )) );
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( cloned->clonePlusChunk( chunk,
+                                              ChunkVersion( 1,
+                                                            0,
+                                                            cloned->getCollVersion().epoch() ),
+                                              &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 15 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 25 )) );
+    }
+
     /**
      * Fixture with single chunk containing:
      * [10->20)
@@ -229,9 +457,9 @@ namespace {
 
         string errMsg;
         const ChunkVersion zeroVersion( 0, 0, OID() );
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinus( chunk,
-                                                                             zeroVersion,
-                                                                             &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinusChunk( chunk,
+                                                                                  zeroVersion,
+                                                                                  &errMsg ) );
 
         ASSERT( errMsg.empty() );
         ASSERT_EQUALS( 0u, cloned->getNumChunks() );
@@ -248,14 +476,120 @@ namespace {
 
         string errMsg;
         ChunkVersion version( 99, 0, OID() );
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinus( chunk,
-                                                                             version,
-                                                                             &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinusChunk( chunk,
+                                                                                  version,
+                                                                                  &errMsg ) );
 
         ASSERT( cloned == NULL );
         ASSERT_FALSE( errMsg.empty() );
     }
 
+    TEST_F(SingleChunkFixture, PlusPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyBelongsToMe(BSON("a" << 15)) );
+        ASSERT( !cloned->keyBelongsToMe(BSON("a" << 25)) );
+        ASSERT( !cloned->keyIsPending(BSON("a" << 15)) );
+        ASSERT( cloned->keyIsPending(BSON("a" << 25)) );
+    }
+
+    TEST_F(SingleChunkFixture, PlusOverlapPendingChunk) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_NOT_EQUALS( errMsg, "" );
+        ASSERT( cloned == NULL );
+    }
+
+    TEST_F(SingleChunkFixture, MinusChunkWithPending) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 35 )) );
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        cloned.reset( cloned->cloneMinusChunk( chunk,
+                                               ChunkVersion( 0,
+                                                             0,
+                                                             cloned->getCollVersion().epoch() ),
+                                               &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 35 )) );
+    }
+
+    TEST_F(SingleChunkFixture, SplitChunkWithPending) {
+
+        string errMsg;
+        ChunkType chunk;
+        scoped_ptr<CollectionMetadata> cloned;
+
+        chunk.setMin( BSON("a" << 20) );
+        chunk.setMax( BSON("a" << 30) );
+
+        cloned.reset( getCollMetadata().clonePlusPending( chunk, &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 35 )) );
+
+        chunk.setMin( BSON("a" << 10) );
+        chunk.setMax( BSON("a" << 20) );
+
+        vector<BSONObj> splitPoints;
+        splitPoints.push_back( BSON("a" << 14) );
+        splitPoints.push_back( BSON("a" << 16) );
+
+        cloned.reset( cloned->cloneSplit( chunk,
+                                          splitPoints,
+                                          ChunkVersion( cloned->getCollVersion().majorVersion() + 1,
+                                                        0,
+                                                        cloned->getCollVersion().epoch() ),
+                                          &errMsg ) );
+
+        ASSERT_EQUALS( errMsg, "" );
+        ASSERT( cloned != NULL );
+
+        ASSERT( cloned->keyIsPending(BSON( "a" << 25 )) );
+        ASSERT( !cloned->keyIsPending(BSON( "a" << 35 )) );
+    }
+
     /**
      * Fixture with single chunk containing:
      * [(min, min)->(max, max))
@@ -386,9 +720,9 @@ namespace {
 
         string errMsg;
         ChunkVersion version( 1, 0, OID() );
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlus( chunk,
-                                                                             version,
-                                                                             &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlusChunk( chunk,
+                                                                                 version,
+                                                                                 &errMsg ) );
 
         ASSERT( errMsg.empty() );
         ASSERT_EQUALS( 2u, getCollMetadata().getNumChunks() );
@@ -410,11 +744,11 @@ namespace {
         chunk.setMax( BSON("a" << 25 << "b" << 0) );
 
         string errMsg;
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlus( chunk,
-                                                                            ChunkVersion( 1,
-                                                                                          0,
-                                                                                          OID() ),
-                                                                            &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().clonePlusChunk( chunk,
+                                                                                 ChunkVersion( 1,
+                                                                                               0,
+                                                                                               OID() ),
+                                                                                 &errMsg ) );
         ASSERT( cloned == NULL );
         ASSERT_FALSE( errMsg.empty() );
         ASSERT_EQUALS( 2u, getCollMetadata().getNumChunks() );
@@ -427,9 +761,9 @@ namespace {
 
         string errMsg;
         ChunkVersion version( 2, 0, OID() );
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinus( chunk,
-                                                                             version,
-                                                                             &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinusChunk( chunk,
+                                                                                  version,
+                                                                                  &errMsg ) );
 
         ASSERT( errMsg.empty() );
         ASSERT_EQUALS( 2u, getCollMetadata().getNumChunks() );
@@ -450,11 +784,11 @@ namespace {
         chunk.setMax( BSON("a" << 28 << "b" << 0) );
 
         string errMsg;
-        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinus( chunk,
-                                                                             ChunkVersion( 1,
-                                                                                           0,
-                                                                                           OID() ),
-                                                                             &errMsg ) );
+        scoped_ptr<CollectionMetadata> cloned( getCollMetadata().cloneMinusChunk( chunk,
+                                                                                  ChunkVersion( 1,
+                                                                                                0,
+                                                                                                OID() ),
+                                                                                  &errMsg ) );
         ASSERT( cloned == NULL );
         ASSERT_FALSE( errMsg.empty() );
         ASSERT_EQUALS( 2u, getCollMetadata().getNumChunks() );
diff --git a/src/mongo/s/d_state.cpp b/src/mongo/s/d_state.cpp
index 1ecd5649a75..7ae236b3faa 100644
--- a/src/mongo/s/d_state.cpp
+++ b/src/mongo/s/d_state.cpp
@@ -172,7 +172,7 @@ namespace mongo {
         chunk.setMax( max );
         string errMsg;
 
-        CollectionMetadataPtr cloned( p->cloneMinus( chunk, version, &errMsg ) );
+        CollectionMetadataPtr cloned( p->cloneMinusChunk( chunk, version, &errMsg ) );
         // uassert to match old behavior, TODO: report errors w/o throwing
         uassert( 16855, errMsg, NULL != cloned.get() );
 
@@ -193,7 +193,7 @@ namespace mongo {
         chunk.setMax( max );
         string errMsg;
 
-        CollectionMetadataPtr cloned( it->second->clonePlus( chunk, version, &errMsg ) );
+        CollectionMetadataPtr cloned( it->second->clonePlusChunk( chunk, version, &errMsg ) );
         // uassert to match old behavior, TODO: report errors w/o throwing
         uassert( 16856, errMsg, NULL != cloned.get() );
 
@@ -312,8 +312,9 @@ namespace mongo {
                                                          currMetadata.get(),
                                                          newMetadataRaw );
 
-        if ( status.code() == ErrorCodes::RemoteChangeDetected ) {
-            version = ChunkVersion( 0, OID() );
+        if ( status.code() == ErrorCodes::RemoteChangeDetected
+                || status.code() == ErrorCodes::NamespaceNotFound ) {
+            version = ChunkVersion( 0, 0, OID() );
             warning() << "did not load new metadata for " << causedBy( status.reason() ) << endl;
             // we loaded something unexpected, the collection may be dropped or dropping
             return false;
diff --git a/src/mongo/s/metadata_loader.cpp b/src/mongo/s/metadata_loader.cpp
index 9eac7b25927..00b83bd4bc9 100644
--- a/src/mongo/s/metadata_loader.cpp
+++ b/src/mongo/s/metadata_loader.cpp
@@ -70,7 +70,7 @@ namespace mongo {
     // MetadataLoader implementation
     //
 
-    MetadataLoader::MetadataLoader( ConnectionString configLoc ) :
+    MetadataLoader::MetadataLoader( const ConnectionString& configLoc ) :
             _configLoc( configLoc )
     {
     }
@@ -81,7 +81,7 @@ namespace mongo {
     Status MetadataLoader::makeCollectionMetadata( const string& ns,
                                                    const string& shard,
                                                    const CollectionMetadata* oldMetadata,
-                                                   CollectionMetadata* metadata )
+                                                   CollectionMetadata* metadata ) const
     {
         Status status = initCollection( ns, shard, metadata );
         if ( !status.isOK() || metadata->getKeyPattern().isEmpty() ) return status;
@@ -90,17 +90,17 @@ namespace mongo {
 
     Status MetadataLoader::initCollection( const string& ns,
                                            const string& shard,
-                                           CollectionMetadata* metadata )
+                                           CollectionMetadata* metadata ) const
     {
         //
         // Bring collection entry from the config server.
         //
 
-        BSONObj collObj;
+        BSONObj collDoc;
         {
             try {
                 ScopedDbConnection conn( _configLoc.toString(), 30 );
-                collObj = conn->findOne( CollectionType::ConfigNS, QUERY(CollectionType::ns()<<ns));
+                collDoc = conn->findOne( CollectionType::ConfigNS, QUERY(CollectionType::ns()<<ns));
                 conn.done();
             }
             catch ( const DBException& e ) {
@@ -114,41 +114,72 @@ namespace mongo {
             }
         }
 
-        CollectionType collDoc;
         string errMsg;
-        if ( !collDoc.parseBSON( collObj, &errMsg ) || !collDoc.isValid( &errMsg ) ) {
+        if ( collDoc.isEmpty() ) {
+
+            errMsg = str::stream() << "could not load metadata, collection " << ns << " not found";
+            warning() << errMsg << endl;
+
+            return Status( ErrorCodes::NamespaceNotFound, errMsg );
+        }
+
+        CollectionType collInfo;
+        if ( !collInfo.parseBSON( collDoc, &errMsg ) || !collInfo.isValid( &errMsg ) ) {
+
+            errMsg = str::stream() << "could not parse metadata for collection " << ns
+                                   << causedBy( errMsg );
+            warning() << errMsg << endl;
+
             return Status( ErrorCodes::FailedToParse, errMsg );
         }
 
-        //
-        // Load or generate default chunks for collection config.
-        //
+        if ( collInfo.isDroppedSet() && collInfo.getDropped() ) {
 
-        if ( collDoc.isKeyPatternSet() && !collDoc.getKeyPattern().isEmpty() ) {
+            errMsg = str::stream() << "could not load metadata, collection " << ns
+                                   << " was dropped";
+            warning() << errMsg << endl;
 
-            metadata->_keyPattern = collDoc.getKeyPattern();
-            metadata->_shardVersion = ChunkVersion( 0, 0, collDoc.getEpoch() );
-            metadata->_collVersion = ChunkVersion( 0, 0, collDoc.getEpoch() );
+            return Status( ErrorCodes::NamespaceNotFound, errMsg );
+        }
+
+        if ( collInfo.isKeyPatternSet() && !collInfo.getKeyPattern().isEmpty() ) {
+
+            // Sharded collection, need to load chunks
+
+            metadata->_keyPattern = collInfo.getKeyPattern();
+            metadata->_shardVersion = ChunkVersion( 0, 0, collInfo.getEpoch() );
+            metadata->_collVersion = ChunkVersion( 0, 0, collInfo.getEpoch() );
 
             return Status::OK();
         }
-        else if ( collDoc.isPrimarySet() && collDoc.getPrimary() == shard ) {
+        else if ( collInfo.isPrimarySet() && collInfo.getPrimary() == shard ) {
 
-            if ( shard == "" ) {
-                warning() << "shard not verified, assuming collection " << ns
-                          << " is unsharded on this shard" << endl;
-            }
+            // A collection with a non-default primary
+
+            // Empty primary field not allowed if set
+            dassert( collInfo.getPrimary() != "" );
 
             metadata->_keyPattern = BSONObj();
-            metadata->_shardVersion = ChunkVersion( 1, 0, collDoc.getEpoch() );
+            metadata->_shardVersion = ChunkVersion( 1, 0, collInfo.getEpoch() );
             metadata->_collVersion = metadata->_shardVersion;
 
             return Status::OK();
         }
         else {
-            errMsg = str::stream() << "collection " << ns << " does not have a shard key "
-                    << "and primary " << ( collDoc.isPrimarySet() ? collDoc.getPrimary() : "" )
-                    << " does not match this shard " << shard;
+
+            // A collection with a primary that doesn't match this shard or is empty, the primary
+            // may have changed before we loaded.
+
+            errMsg = // br
+                    str::stream() << "collection " << ns << " does not have a shard key "
+                                  << "and primary "
+                                  << ( collInfo.isPrimarySet() ? collInfo.getPrimary() : "" )
+                                  << " does not match this shard " << shard;
+
+            warning() << errMsg << endl;
+
+            metadata->_collVersion = ChunkVersion( 0, 0, OID() );
+
             return Status( ErrorCodes::RemoteChangeDetected, errMsg );
         }
     }
@@ -156,7 +187,7 @@ namespace mongo {
     Status MetadataLoader::initChunks( const string& ns,
                                        const string& shard,
                                        const CollectionMetadata* oldMetadata,
-                                       CollectionMetadata* metadata )
+                                       CollectionMetadata* metadata ) const
     {
         map<string, ChunkVersion> versionMap;
         OID epoch = metadata->getCollVersion().epoch();
@@ -183,6 +214,10 @@ namespace mongo {
                               << " and " << metadata->_chunksMap.size() << " chunks" << endl;
             }
         }
+        else {
+            // Preserve the epoch
+            versionMap[shard] = metadata->_shardVersion;
+        }
 
         // Exposes the new metadata's range map and version to the "differ," who
         // would ultimately be responsible of filling them up.
@@ -197,50 +232,70 @@ namespace mongo {
                                                            differ.configDiffQuery() );
 
             if ( !cursor.get() ) {
-                metadata->_collVersion = ChunkVersion();
+
+                // Make our metadata invalid
+                metadata->_collVersion = ChunkVersion( 0, 0, OID() );
                 metadata->_chunksMap.clear();
                 conn.done();
+
                 return Status( ErrorCodes::HostUnreachable,
                                "problem opening chunk metadata cursor" );
             }
 
-            // Diff tracker should *always* find at least one chunk if this shard owns a chunk.
+            //
+            // The diff tracker should always find at least one chunk (the highest chunk we saw
+            // last time).  If not, something has changed on the config server (potentially between
+            // when we read the collection data and when we read the chunks data).
+            //
+
             int diffsApplied = differ.calculateConfigDiff( *cursor );
             if ( diffsApplied > 0 ) {
 
+                // Chunks found, return ok
+
                 LOG(2) << "loaded " << diffsApplied << " chunks into new metadata for " << ns
                            << " with version " << metadata->_collVersion << endl;
 
                 metadata->_shardVersion = versionMap[shard];
                 metadata->fillRanges();
                 conn.done();
+
                 return Status::OK();
             }
             else if ( diffsApplied == 0 ) {
 
-                warning() << "no chunks found when reloading " << ns << ", previous version was "
-                          << metadata->_collVersion.toString() << endl;
+                // No chunks found, something changed or we're confused
+
+                string errMsg = // br
+                        str::stream() << "no chunks found when reloading " << ns
+                                      << ", previous version was "
+                                      << metadata->_collVersion.toString();
+
+                warning() << errMsg << endl;
 
                 metadata->_collVersion = ChunkVersion( 0, 0, OID() );
                 metadata->_chunksMap.clear();
                 conn.done();
-                return Status::OK();
+
+                return Status( ErrorCodes::RemoteChangeDetected, errMsg );
             }
             else {
 
-                // TODO: make this impossible by making sure we don't migrate / split on this
-                // shard during the reload.  No chunks were found for the ns.
+                // Invalid chunks found, our epoch may have changed because we dropped/recreated
+                // the collection.
 
-                string errMsg = str::stream() << "invalid chunks found when reloading " << ns
-                                              << ", previous version was "
-                                              << metadata->_collVersion.toString()
-                                              << ", this should be rare";
+                string errMsg = // br
+                        str::stream() << "invalid chunks found when reloading " << ns
+                                      << ", previous version was "
+                                      << metadata->_collVersion.toString()
+                                      << ", this should be rare";
 
                 warning() << errMsg << endl;
 
                 metadata->_collVersion = ChunkVersion( 0, 0, OID() );
                 metadata->_chunksMap.clear();
                 conn.done();
+
                 return Status( ErrorCodes::RemoteChangeDetected, errMsg );
             }
         }
diff --git a/src/mongo/s/metadata_loader.h b/src/mongo/s/metadata_loader.h
index ebefe9614a3..bce95b5d675 100644
--- a/src/mongo/s/metadata_loader.h
+++ b/src/mongo/s/metadata_loader.h
@@ -47,7 +47,7 @@ namespace mongo {
          * that we make no restrictions about which connection string that is, including
          * CUSTOM, which we rely on in testing.
          */
-        explicit MetadataLoader( ConnectionString configLoc );
+        explicit MetadataLoader( const ConnectionString& configLoc );
 
         ~MetadataLoader();
 
@@ -63,13 +63,14 @@ namespace mongo {
          * Abnormal:
          * @return FailedToParse if there was an error parsing the remote config data
          * Normal:
+         * @return NamespaceNotFound if the collection no longer exists
          * @return HostUnreachable if there was an error contacting the config servers
          * @return RemoteChangeDetected if the data loaded was modified by another operation
          */
         Status makeCollectionMetadata( const string& ns,
                                        const string& shard,
                                        const CollectionMetadata* oldMetadata,
-                                       CollectionMetadata* metadata );
+                                       CollectionMetadata* metadata ) const;
 
     private:
         ConnectionString _configLoc;
@@ -79,6 +80,7 @@ namespace mongo {
          * information, not including chunks.
          *
          * If information about the collection can be accessed or is invalid, returns:
+         * @return NamespaceNotFound if the collection no longer exists
          * @return FailedToParse if there was an error parsing the remote config data
          * @return HostUnreachable if there was an error contacting the config servers
          * @return RemoteChangeDetected if the collection doc loaded is unexpectedly different
@@ -86,7 +88,7 @@ namespace mongo {
          */
         Status initCollection( const string& ns,
                                const string& shard,
-                               CollectionMetadata* metadata );
+                               CollectionMetadata* metadata ) const;
 
         /**
          * Returns OK and fills in the chunk state of 'metadata' to portray the chunks of the
@@ -101,7 +103,7 @@ namespace mongo {
         Status initChunks( const string& ns,
                            const string& shard,
                            const CollectionMetadata* oldMetadata,
-                           CollectionMetadata* metadata );
+                           CollectionMetadata* metadata ) const;
     };
 
 } // namespace mongo
diff --git a/src/mongo/s/metadata_loader_test.cpp b/src/mongo/s/metadata_loader_test.cpp
index 2d8724441f9..6072fc7b741 100644
--- a/src/mongo/s/metadata_loader_test.cpp
+++ b/src/mongo/s/metadata_loader_test.cpp
@@ -56,12 +56,14 @@ namespace {
     using std::vector;
 
     const std::string CONFIG_HOST_PORT = "$dummy_config:27017";
+    const ConnectionString CONFIG_LOC( CONFIG_HOST_PORT );
 
     // TODO: Test config server down
     // TODO: Test read of chunks with new epoch
     // TODO: Test that you can properly load config using format with deprecated fields?
 
     TEST(MetadataLoader, DroppedColl) {
+
         MockRemoteDBServer dummyConfig( CONFIG_HOST_PORT );
         mongo::ConnectionString::setConnectionHook( MockConnRegistry::get()->getConnStrHook() );
         MockConnRegistry::get()->addServer( &dummyConfig );
@@ -77,11 +79,91 @@ namespace {
 
         dummyConfig.insert( CollectionType::ConfigNS, collInfo.toBSON() );
 
-        dummyConfig.insert( ChunkType::ConfigNS, BSONObj() );
+        MetadataLoader loader( CONFIG_LOC );
 
-        ConnectionString confServerStr( CONFIG_HOST_PORT );
-        ConnectionString configLoc( confServerStr );
-        MetadataLoader loader( configLoc );
+        string errmsg;
+        CollectionMetadata metadata;
+        Status status = loader.makeCollectionMetadata( "test.foo", // br
+                                                       "shard0000",
+                                                       NULL, /* no old metadata */
+                                                       &metadata );
+
+        ASSERT_EQUALS( status.code(), ErrorCodes::NamespaceNotFound );
+
+        MockConnRegistry::get()->clear();
+        ScopedDbConnection::clearPool();
+    }
+
+    TEST(MetadataLoader, EmptyColl) {
+
+        MockRemoteDBServer dummyConfig( CONFIG_HOST_PORT );
+        mongo::ConnectionString::setConnectionHook( MockConnRegistry::get()->getConnStrHook() );
+        MockConnRegistry::get()->addServer( &dummyConfig );
+
+        MetadataLoader loader( CONFIG_LOC );
+
+        string errmsg;
+        CollectionMetadata metadata;
+        Status status = loader.makeCollectionMetadata( "test.foo", // br
+                                                       "shard0000",
+                                                       NULL, /* no old metadata */
+                                                       &metadata );
+
+        ASSERT_EQUALS( status.code(), ErrorCodes::NamespaceNotFound );
+
+        MockConnRegistry::get()->clear();
+        ScopedDbConnection::clearPool();
+    }
+
+    TEST(MetadataLoader, BadColl) {
+
+        MockRemoteDBServer dummyConfig( CONFIG_HOST_PORT );
+        mongo::ConnectionString::setConnectionHook( MockConnRegistry::get()->getConnStrHook() );
+        MockConnRegistry::get()->addServer( &dummyConfig );
+
+        dummyConfig.insert( CollectionType::ConfigNS, BSON( CollectionType::ns("test.foo") ) );
+
+        MetadataLoader loader( CONFIG_LOC );
+
+        string errmsg;
+        CollectionMetadata metadata;
+        Status status = loader.makeCollectionMetadata( "test.foo", // br
+                                                       "shard0000",
+                                                       NULL, /* no old metadata */
+                                                       &metadata );
+
+        ASSERT_EQUALS( status.code(), ErrorCodes::FailedToParse );
+
+        MockConnRegistry::get()->clear();
+        ScopedDbConnection::clearPool();
+    }
+
+
+    TEST(MetadataLoader, BadChunk) {
+
+        MockRemoteDBServer dummyConfig( CONFIG_HOST_PORT );
+        mongo::ConnectionString::setConnectionHook( MockConnRegistry::get()->getConnStrHook() );
+        MockConnRegistry::get()->addServer( &dummyConfig );
+
+        CollectionType collInfo;
+        collInfo.setNS( "test.foo");
+        collInfo.setUpdatedAt( 0 );
+        collInfo.setKeyPattern( BSON("a" << 1) );
+        collInfo.setEpoch( OID() );
+
+        string errMsg;
+        ASSERT( collInfo.isValid( &errMsg ) );
+
+        dummyConfig.insert( CollectionType::ConfigNS, collInfo.toBSON() );
+
+        ChunkType chunkInfo;
+        chunkInfo.setNS( "test.foo");
+        chunkInfo.setVersion(ChunkVersion(1, 0, OID()));
+        ASSERT( !chunkInfo.isValid( &errMsg ) );
+
+        dummyConfig.insert( ChunkType::ConfigNS, chunkInfo.toBSON() );
+
+        MetadataLoader loader( CONFIG_LOC );
 
         string errmsg;
         CollectionMetadata metadata;
@@ -90,6 +172,8 @@ namespace {
                                                        NULL, /* no old metadata */
                                                        &metadata );
 
+        // For now, since the differ doesn't have parsing errors, we get this kind of status
+        // TODO: Make the differ do parse errors
         ASSERT_EQUALS( status.code(), ErrorCodes::RemoteChangeDetected );
 
         MockConnRegistry::get()->clear();
@@ -123,23 +207,66 @@ namespace {
     };
 
     TEST_F(NoChunkFixture, CheckNumChunk) {
-        ConnectionString confServerStr( CONFIG_HOST_PORT );
-        ConnectionString configLoc( confServerStr );
-        MetadataLoader loader( configLoc );
+
+        MetadataLoader loader( CONFIG_LOC );
 
         CollectionMetadata metadata;
         Status status = loader.makeCollectionMetadata( "test.foo", // br
                                                        "shard0000",
                                                        NULL, /* no old metadata */
                                                        &metadata );
-        ASSERT_TRUE( status.isOK() );
-        ASSERT_EQUALS( 0U, metadata.getNumChunks() );
+
+        ASSERT_EQUALS( status.code(), ErrorCodes::RemoteChangeDetected );
     }
 
-    TEST_F(NoChunkFixture, VersionIsZero) {
+    class NoChunkHereFixture : public mongo::unittest::Test {
+    protected:
+        void setUp() {
+            _dummyConfig.reset( new MockRemoteDBServer( CONFIG_HOST_PORT ) );
+            mongo::ConnectionString::setConnectionHook( MockConnRegistry::get()->getConnStrHook() );
+            MockConnRegistry::get()->addServer( _dummyConfig.get() );
+
+            OID epoch = OID::gen();
+
+            CollectionType collType;
+            collType.setNS( "test.foo" );
+            collType.setKeyPattern( BSON("a" << 1) );
+            collType.setUnique( false );
+            collType.setUpdatedAt( 1ULL );
+            collType.setEpoch( epoch );
+            string errMsg;
+            ASSERT( collType.isValid( &errMsg ) );
+
+            _dummyConfig->insert( CollectionType::ConfigNS, collType.toBSON() );
+
+            // Need a chunk on another shard, otherwise the chunks are invalid in general and we
+            // can't load metadata
+            ChunkType chunkType;
+            chunkType.setNS( "test.foo");
+            chunkType.setShard( "shard0001" );
+            chunkType.setMin( BSON( "a" << MINKEY ) );
+            chunkType.setMax( BSON( "a" << MAXKEY ) );
+            chunkType.setVersion( ChunkVersion( 1, 0, epoch ) );
+            chunkType.setName( OID::gen().toString() );
+            ASSERT( chunkType.isValid( &errMsg ) );
+
+            _dummyConfig->insert( ChunkType::ConfigNS, chunkType.toBSON() );
+        }
+
+        void tearDown() {
+            MockConnRegistry::get()->clear();
+            ScopedDbConnection::clearPool();
+        }
+
+    private:
+        scoped_ptr<MockRemoteDBServer> _dummyConfig;
+    };
+
+    TEST_F(NoChunkHereFixture, CheckNumChunk) {
         ConnectionString confServerStr( CONFIG_HOST_PORT );
         ConnectionString configLoc( confServerStr );
         MetadataLoader loader( configLoc );
+
         CollectionMetadata metadata;
         Status status = loader.makeCollectionMetadata( "test.foo", // br
                                                        "shard0000",
@@ -147,7 +274,11 @@ namespace {
                                                        &metadata );
 
         ASSERT( status.isOK() );
-        ASSERT_EQUALS( 0U, metadata.getShardVersion().toLong() );
+        ASSERT_EQUALS( 0U, metadata.getNumChunks() );
+        ASSERT_EQUALS( 1, metadata.getCollVersion().majorVersion() );
+        ASSERT_EQUALS( 0, metadata.getShardVersion().majorVersion() );
+        ASSERT_NOT_EQUALS( OID(), metadata.getCollVersion().epoch() );
+        ASSERT_NOT_EQUALS( OID(), metadata.getShardVersion().epoch() );
     }
 
     class ConfigServerFixture : public mongo::unittest::Test {
@@ -267,12 +398,7 @@ namespace {
                                                        NULL, /* no old metadata */
                                                        &metadata );
 
-        ASSERT( status.isOK() );
-
-        ChunkVersion versionZero( 0, 0, OID() );
-        ASSERT_EQUALS( versionZero.toLong(), metadata.getCollVersion().toLong() );
-        ASSERT_EQUALS( versionZero.toLong(), metadata.getShardVersion().toLong() );
-        ASSERT_EQUALS( 0U, metadata.getNumChunks() );
+        ASSERT_EQUALS( status.code(), ErrorCodes::RemoteChangeDetected );
     }
 
 #if 0
diff --git a/src/mongo/s/range_arithmetic.cpp b/src/mongo/s/range_arithmetic.cpp
index d0ecd273bb4..051cdea4362 100644
--- a/src/mongo/s/range_arithmetic.cpp
+++ b/src/mongo/s/range_arithmetic.cpp
@@ -17,27 +17,95 @@
 #include "mongo/s/range_arithmetic.h"
 
 namespace mongo {
-    bool rangeContains(const BSONObj& inclusiveLower,
-                       const BSONObj& exclusiveUpper,
-                       const BSONObj& point) {
-        return point.woCompare(inclusiveLower) >= 0 &&
-                point.woCompare(exclusiveUpper) < 0;
+
+    using std::make_pair;
+    using std::pair;
+    using std::string;
+    using std::stringstream;
+
+    bool rangeContains( const BSONObj& inclusiveLower,
+                        const BSONObj& exclusiveUpper,
+                        const BSONObj& point )
+    {
+        return point.woCompare( inclusiveLower ) >= 0 && point.woCompare( exclusiveUpper ) < 0;
     }
 
-    bool rangeOverlaps(const BSONObj& inclusiveLower1,
-                       const BSONObj& exclusiveUpper1,
-                       const BSONObj& inclusiveLower2,
-                       const BSONObj& exclusiveUpper2) {
-        return (exclusiveUpper1.woCompare(inclusiveLower2) > 0) &&
-                (exclusiveUpper2.woCompare(inclusiveLower1) > 0);
+    bool rangeOverlaps( const BSONObj& inclusiveLower1,
+                        const BSONObj& exclusiveUpper1,
+                        const BSONObj& inclusiveLower2,
+                        const BSONObj& exclusiveUpper2 )
+    {
+        return ( exclusiveUpper1.woCompare( inclusiveLower2 ) > 0 )
+            && ( exclusiveUpper2.woCompare( inclusiveLower1 ) > 0 );
     }
 
-    int compareRanges(const BSONObj& rangeMin1,
-                      const BSONObj& rangeMax1,
-                      const BSONObj& rangeMin2,
-                      const BSONObj& rangeMax2) {
-        const int minCmp = rangeMin1.woCompare(rangeMin2);
-        if (minCmp != 0) return minCmp;
-        return rangeMax1.woCompare(rangeMax2);
+    int compareRanges( const BSONObj& rangeMin1,
+                       const BSONObj& rangeMax1,
+                       const BSONObj& rangeMin2,
+                       const BSONObj& rangeMax2 )
+    {
+        const int minCmp = rangeMin1.woCompare( rangeMin2 );
+        if ( minCmp != 0 ) return minCmp;
+        return rangeMax1.woCompare( rangeMax2 );
     }
+
+    // Represents the start and end of an overlap of a tested range
+    typedef pair<RangeMap::const_iterator, RangeMap::const_iterator> OverlapBounds;
+
+    // Internal-only, shared functionality
+    OverlapBounds rangeMapOverlapBounds( const RangeMap& ranges,
+                                         const BSONObj& inclusiveLower,
+                                         const BSONObj& exclusiveUpper ) {
+
+        // Returns the first chunk with a min key that is >= lower bound - the previous chunk
+        // might overlap.
+        RangeMap::const_iterator low = ranges.lower_bound( inclusiveLower );
+
+        // See if the previous chunk overlaps our range, not clear from just min key
+        if ( low != ranges.begin() ) {
+
+            RangeMap::const_iterator next = low;
+            --low;
+
+            // If the previous range's max value is lte our min value
+            if ( low->second.woCompare( inclusiveLower ) < 1 ) {
+                low = next;
+            }
+        }
+
+        // Returns the first chunk with a max key that is >= upper bound - implies the
+        // chunk does not overlap upper bound
+        RangeMap::const_iterator high = ranges.lower_bound( exclusiveUpper );
+
+        return OverlapBounds( low, high );
+    }
+
+    void getRangeMapOverlap( const RangeMap& ranges,
+                             const BSONObj& inclusiveLower,
+                             const BSONObj& exclusiveUpper,
+                             RangeVector* overlap ) {
+        overlap->clear();
+        OverlapBounds bounds = rangeMapOverlapBounds( ranges, inclusiveLower, exclusiveUpper );
+        for ( RangeMap::const_iterator it = bounds.first; it != bounds.second; ++it ) {
+            overlap->push_back( make_pair( it->first, it->second ) );
+        }
+    }
+
+    bool rangeMapOverlaps( const RangeMap& ranges,
+                           const BSONObj& inclusiveLower,
+                           const BSONObj& exclusiveUpper ) {
+        OverlapBounds bounds = rangeMapOverlapBounds( ranges, inclusiveLower, exclusiveUpper );
+        return bounds.first != bounds.second;
+    }
+
+    bool rangeMapContains( const RangeMap& ranges,
+                           const BSONObj& inclusiveLower,
+                           const BSONObj& exclusiveUpper ) {
+        OverlapBounds bounds = rangeMapOverlapBounds( ranges, inclusiveLower, exclusiveUpper );
+        if ( bounds.first == ranges.end() ) return false;
+
+        return bounds.first->first.woCompare( inclusiveLower ) == 0
+            && bounds.first->second.woCompare( exclusiveUpper ) == 0;
+    }
+
 }
diff --git a/src/mongo/s/range_arithmetic.h b/src/mongo/s/range_arithmetic.h
index b83dca52319..24f257e2718 100644
--- a/src/mongo/s/range_arithmetic.h
+++ b/src/mongo/s/range_arithmetic.h
@@ -17,6 +17,8 @@
 #pragma once
 
 #include <string>
+#include <map>
+#include <vector>
 
 #include "mongo/db/jsobj.h"
 
@@ -88,4 +90,43 @@ namespace mongo {
                        const BSONObj& rangeMin2,
                        const BSONObj& rangeMax2 );
 
+    /**
+     * A RangeMap is a mapping of a BSON range from lower->upper (lower maps to upper), using
+     * standard BSON woCompare.  Upper bound is exclusive.
+     *
+     * NOTE: For overlap testing to work correctly, there may be no overlaps present in the map
+     * itself.
+     */
+    typedef map<BSONObj, BSONObj, BSONObjCmp> RangeMap;
+
+    /**
+     * A RangeVector is a list of [lower,upper) ranges.
+     */
+    typedef vector<pair<BSONObj,BSONObj> > RangeVector;
+
+    /**
+     * Returns the overlap of a range [inclusiveLower, exclusiveUpper) with the provided range map
+     * as a vector of ranges from the map.
+     */
+    void getRangeMapOverlap( const RangeMap& ranges,
+                             const BSONObj& inclusiveLower,
+                             const BSONObj& exclusiveUpper,
+                             RangeVector* vector );
+
+    /**
+     * Returns true if the provided range map has ranges which overlap the provided range
+     * [inclusiveLower, exclusiveUpper).
+     */
+    bool rangeMapOverlaps( const RangeMap& ranges,
+                           const BSONObj& inclusiveLower,
+                           const BSONObj& exclusiveUpper );
+
+    /**
+     * Returns true if the provided range map exactly contains the provided range
+     * [inclusiveLower, exclusiveUpper).
+     */
+    bool rangeMapContains( const RangeMap& ranges,
+                           const BSONObj& inclusiveLower,
+                           const BSONObj& exclusiveUpper );
+
 }
diff --git a/src/mongo/s/range_arithmetic_test.cpp b/src/mongo/s/range_arithmetic_test.cpp
index e0a8af77dc9..37d846d5b34 100644
--- a/src/mongo/s/range_arithmetic_test.cpp
+++ b/src/mongo/s/range_arithmetic_test.cpp
@@ -18,7 +18,12 @@
 #include "mongo/unittest/unittest.h"
 
 namespace {
+
     using mongo::rangeOverlaps;
+    using mongo::rangeMapOverlaps;
+    using mongo::RangeMap;
+    using mongo::RangeVector;
+    using std::make_pair;
 
     TEST(BSONRange, SmallerLowerRangeNonSubset) {
         ASSERT_TRUE(rangeOverlaps(BSON("x" << 100), BSON("x" << 200),
@@ -55,4 +60,79 @@ namespace {
         ASSERT_TRUE(rangeOverlaps(BSON("x" << 100), BSON("x" << 200),
                                   BSON("x" << 100), BSON("x" << 200)));
     }
+
+    TEST(RangeMap, RangeMapOverlap) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        rangeMap.insert( make_pair( BSON( "x" << 200 ), BSON( "x" << 300 ) ) );
+        rangeMap.insert( make_pair( BSON( "x" << 300 ), BSON( "x" << 400 ) ) );
+
+        RangeVector overlap;
+        getRangeMapOverlap( rangeMap, BSON( "x" << 50 ), BSON( "x" << 350 ), &overlap );
+
+        ASSERT( !overlap.empty() );
+        ASSERT_EQUALS( overlap.size(), 3u );
+    }
+
+    TEST(RangeMap, RangeMapOverlapPartial) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        rangeMap.insert( make_pair( BSON( "x" << 200 ), BSON( "x" << 300 ) ) );
+
+        RangeVector overlap;
+        getRangeMapOverlap( rangeMap, BSON( "x" << 150 ), BSON( "x" << 250 ), &overlap );
+
+        ASSERT( !overlap.empty() );
+        ASSERT_EQUALS( overlap.size(), 2u );
+    }
+
+    TEST(RangeMap, RangeMapOverlapInner) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+
+        RangeVector overlap;
+        getRangeMapOverlap( rangeMap, BSON( "x" << 125 ), BSON( "x" << 150 ), &overlap );
+
+        ASSERT( !overlap.empty() );
+        ASSERT_EQUALS( overlap.size(), 1u );
+    }
+
+    TEST(RangeMap, RangeMapNoOverlap) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        rangeMap.insert( make_pair( BSON( "x" << 300 ), BSON( "x" << 400 ) ) );
+
+        RangeVector overlap;
+        getRangeMapOverlap( rangeMap, BSON( "x" << 200 ), BSON( "x" << 300 ), &overlap );
+
+        ASSERT( overlap.empty() );
+    }
+
+    TEST(RangeMap, RangeMapOverlaps) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+
+        ASSERT( rangeMapOverlaps( rangeMap, BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        ASSERT( rangeMapOverlaps( rangeMap, BSON( "x" << 99 ), BSON( "x" << 200 ) ) );
+        ASSERT( rangeMapOverlaps( rangeMap, BSON( "x" << 100 ), BSON( "x" << 201 ) ) );
+        ASSERT( rangeMapOverlaps( rangeMap, BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        ASSERT( !rangeMapOverlaps( rangeMap, BSON( "x" << 99 ), BSON( "x" << 100 ) ) );
+        ASSERT( !rangeMapOverlaps( rangeMap, BSON( "x" << 200 ), BSON( "x" << 201 ) ) );
+    }
+
+    TEST(RangeMap, RangeMapContains) {
+
+        RangeMap rangeMap;
+        rangeMap.insert( make_pair( BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+
+        ASSERT( rangeMapContains( rangeMap, BSON( "x" << 100 ), BSON( "x" << 200 ) ) );
+        ASSERT( !rangeMapContains( rangeMap, BSON( "x" << 99 ), BSON( "x" << 200 ) ) );
+        ASSERT( !rangeMapContains( rangeMap, BSON( "x" << 100 ), BSON( "x" << 201 ) ) );
+    }
+
 }
diff --git a/src/mongo/s/type_collection.h b/src/mongo/s/type_collection.h
index c8bcafc304c..c68e4ddcc89 100644
--- a/src/mongo/s/type_collection.h
+++ b/src/mongo/s/type_collection.h
@@ -179,7 +179,7 @@ namespace mongo {
                 return primary.getDefault();
             }
         }
-        void setKeyPattern(BSONObj& keyPattern) {
+        void setKeyPattern(const BSONObj& keyPattern) {
             _keyPattern = keyPattern.getOwned();
             _isKeyPatternSet = true;
         }