1 files changed, 295 insertions, 326 deletions

diff --git a/src/mongo/db/storage/kv/kv_catalog.cpp b/src/mongo/db/storage/kv/kv_catalog.cpp
index b24cc705226..df0a39faeee 100644
--- a/src/mongo/db/storage/kv/kv_catalog.cpp
+++ b/src/mongo/db/storage/kv/kv_catalog.cpp
@@ -45,398 +45,367 @@
 
 namespace mongo {
 namespace {
-    // This is a global resource, which protects accesses to the catalog metadata (instance-wide).
-    // It is never used with KVEngines that support doc-level locking so this should never conflict
-    // with anything else.
-    //
-    // NOTE: Must be locked *before* _identLock.
-    const ResourceId resourceIdCatalogMetadata(RESOURCE_METADATA, 1ULL);
+// This is a global resource, which protects accesses to the catalog metadata (instance-wide).
+// It is never used with KVEngines that support doc-level locking so this should never conflict
+// with anything else.
+//
+// NOTE: Must be locked *before* _identLock.
+const ResourceId resourceIdCatalogMetadata(RESOURCE_METADATA, 1ULL);
 }
 
-    using std::unique_ptr;
-    using std::string;
+using std::unique_ptr;
+using std::string;
 
-    class KVCatalog::AddIdentChange : public RecoveryUnit::Change {
-    public:
-        AddIdentChange(KVCatalog* catalog, StringData ident)
-            :_catalog(catalog), _ident(ident.toString())
-        {}
+class KVCatalog::AddIdentChange : public RecoveryUnit::Change {
+public:
+    AddIdentChange(KVCatalog* catalog, StringData ident)
+        : _catalog(catalog), _ident(ident.toString()) {}
 
-        virtual void commit() {}
-        virtual void rollback() {
-            stdx::lock_guard<stdx::mutex> lk(_catalog->_identsLock);
-            _catalog->_idents.erase(_ident);
-        }
+    virtual void commit() {}
+    virtual void rollback() {
+        stdx::lock_guard<stdx::mutex> lk(_catalog->_identsLock);
+        _catalog->_idents.erase(_ident);
+    }
 
-        KVCatalog* const _catalog;
-        const std::string _ident;
-    };
+    KVCatalog* const _catalog;
+    const std::string _ident;
+};
 
-    class KVCatalog::RemoveIdentChange : public RecoveryUnit::Change {
-    public:
-        RemoveIdentChange(KVCatalog* catalog, StringData ident, const Entry& entry)
-            :_catalog(catalog), _ident(ident.toString()), _entry(entry)
-        {}
+class KVCatalog::RemoveIdentChange : public RecoveryUnit::Change {
+public:
+    RemoveIdentChange(KVCatalog* catalog, StringData ident, const Entry& entry)
+        : _catalog(catalog), _ident(ident.toString()), _entry(entry) {}
 
-        virtual void commit() {}
-        virtual void rollback() {
-            stdx::lock_guard<stdx::mutex> lk(_catalog->_identsLock);
-            _catalog->_idents[_ident] = _entry;
-        }
+    virtual void commit() {}
+    virtual void rollback() {
+        stdx::lock_guard<stdx::mutex> lk(_catalog->_identsLock);
+        _catalog->_idents[_ident] = _entry;
+    }
+
+    KVCatalog* const _catalog;
+    const std::string _ident;
+    const Entry _entry;
+};
+
+KVCatalog::KVCatalog(RecordStore* rs,
+                     bool isRsThreadSafe,
+                     bool directoryPerDb,
+                     bool directoryForIndexes)
+    : _rs(rs),
+      _isRsThreadSafe(isRsThreadSafe),
+      _directoryPerDb(directoryPerDb),
+      _directoryForIndexes(directoryForIndexes),
+      _rand(_newRand()) {}
+
+KVCatalog::~KVCatalog() {
+    _rs = NULL;
+}
+
+std::string KVCatalog::_newRand() {
+    return str::stream() << std::unique_ptr<SecureRandom>(SecureRandom::create())->nextInt64();
+}
 
-        KVCatalog* const _catalog;
-        const std::string _ident;
-        const Entry _entry;
-    };
-
-    KVCatalog::KVCatalog( RecordStore* rs,
-                          bool isRsThreadSafe,
-                          bool directoryPerDb,
-                          bool directoryForIndexes )
-        : _rs( rs )
-        , _isRsThreadSafe(isRsThreadSafe)
-        , _directoryPerDb(directoryPerDb)
-        , _directoryForIndexes(directoryForIndexes)
-        , _rand(_newRand())
-    {}
-
-    KVCatalog::~KVCatalog() {
-        _rs = NULL;
+bool KVCatalog::_hasEntryCollidingWithRand() const {
+    // Only called from init() so don't need to lock.
+    for (NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it) {
+        if (StringData(it->first).endsWith(_rand))
+            return true;
     }
+    return false;
+}
 
-    std::string KVCatalog::_newRand() {
-        return str::stream()
-            << std::unique_ptr<SecureRandom>(SecureRandom::create())->nextInt64();
+std::string KVCatalog::_newUniqueIdent(StringData ns, const char* kind) {
+    // If this changes to not put _rand at the end, _hasEntryCollidingWithRand will need fixing.
+    StringBuilder buf;
+    if (_directoryPerDb) {
+        buf << NamespaceString::escapeDbName(nsToDatabaseSubstring(ns)) << '/';
     }
+    buf << kind;
+    buf << (_directoryForIndexes ? '/' : '-');
+    buf << _next.fetchAndAdd(1) << '-' << _rand;
+    return buf.str();
+}
 
-    bool KVCatalog::_hasEntryCollidingWithRand() const {
-        // Only called from init() so don't need to lock.
-        for (NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it) {
-            if (StringData(it->first).endsWith(_rand))
-                return true;
-        }
-        return false;
+void KVCatalog::init(OperationContext* opCtx) {
+    // No locking needed since called single threaded.
+    auto cursor = _rs->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        BSONObj obj = record->data.releaseToBson();
+
+        // No rollback since this is just loading already committed data.
+        string ns = obj["ns"].String();
+        string ident = obj["ident"].String();
+        _idents[ns] = Entry(ident, record->id);
     }
 
-    std::string KVCatalog::_newUniqueIdent(StringData ns, const char* kind) {
-        // If this changes to not put _rand at the end, _hasEntryCollidingWithRand will need fixing.
-        StringBuilder buf;
-        if ( _directoryPerDb ) {
-            buf << NamespaceString::escapeDbName( nsToDatabaseSubstring( ns ) ) << '/';
-        }
-        buf << kind;
-        buf << ( _directoryForIndexes ? '/' : '-' );
-        buf << _next.fetchAndAdd(1) << '-' << _rand;
-        return buf.str();
+    // In the unlikely event that we have used this _rand before generate a new one.
+    while (_hasEntryCollidingWithRand()) {
+        _rand = _newRand();
     }
+}
 
-    void KVCatalog::init( OperationContext* opCtx ) {
-        // No locking needed since called single threaded.
-        auto cursor = _rs->getCursor(opCtx);
-        while (auto record = cursor->next()) {
-            BSONObj obj = record->data.releaseToBson();
+void KVCatalog::getAllCollections(std::vector<std::string>* out) const {
+    stdx::lock_guard<stdx::mutex> lk(_identsLock);
+    for (NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it) {
+        out->push_back(it->first);
+    }
+}
 
-            // No rollback since this is just loading already committed data.
-            string ns = obj["ns"].String();
-            string ident = obj["ident"].String();
-            _idents[ns] = Entry(ident, record->id);
-        }
+Status KVCatalog::newCollection(OperationContext* opCtx,
+                                StringData ns,
+                                const CollectionOptions& options) {
+    invariant(opCtx->lockState() == NULL ||
+              opCtx->lockState()->isDbLockedForMode(nsToDatabaseSubstring(ns), MODE_X));
 
-        // In the unlikely event that we have used this _rand before generate a new one.
-        while (_hasEntryCollidingWithRand()) {
-            _rand = _newRand();
-        }
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
     }
 
-    void KVCatalog::getAllCollections( std::vector<std::string>* out ) const {
-        stdx::lock_guard<stdx::mutex> lk( _identsLock );
-        for ( NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it ) {
-            out->push_back( it->first );
-        }
+    const string ident = _newUniqueIdent(ns, "collection");
+
+    stdx::lock_guard<stdx::mutex> lk(_identsLock);
+    Entry& old = _idents[ns.toString()];
+    if (!old.ident.empty()) {
+        return Status(ErrorCodes::NamespaceExists, "collection already exists");
     }
 
-    Status KVCatalog::newCollection( OperationContext* opCtx,
-                                     StringData ns,
-                                     const CollectionOptions& options ) {
-        invariant( opCtx->lockState() == NULL ||
-                   opCtx->lockState()->isDbLockedForMode( nsToDatabaseSubstring(ns), MODE_X ) );
-
-        std::unique_ptr<Lock::ResourceLock> rLk;
-        if (!_isRsThreadSafe && opCtx->lockState()) {
-            rLk.reset(new Lock::ResourceLock(opCtx->lockState(),
-                                             resourceIdCatalogMetadata,
-                                             MODE_X));
-        }
+    opCtx->recoveryUnit()->registerChange(new AddIdentChange(this, ns));
 
-        const string ident = _newUniqueIdent(ns, "collection");
+    BSONObj obj;
+    {
+        BSONObjBuilder b;
+        b.append("ns", ns);
+        b.append("ident", ident);
+        BSONCollectionCatalogEntry::MetaData md;
+        md.ns = ns.toString();
+        md.options = options;
+        b.append("md", md.toBSON());
+        obj = b.obj();
+    }
 
-        stdx::lock_guard<stdx::mutex> lk( _identsLock );
-        Entry& old = _idents[ns.toString()];
-        if ( !old.ident.empty() ) {
-            return Status( ErrorCodes::NamespaceExists, "collection already exists" );
-        }
+    StatusWith<RecordId> res = _rs->insertRecord(opCtx, obj.objdata(), obj.objsize(), false);
+    if (!res.isOK())
+        return res.getStatus();
 
-        opCtx->recoveryUnit()->registerChange(new AddIdentChange(this, ns));
-
-        BSONObj obj;
-        {
-            BSONObjBuilder b;
-            b.append( "ns", ns );
-            b.append( "ident", ident );
-            BSONCollectionCatalogEntry::MetaData md;
-            md.ns = ns.toString();
-            md.options = options;
-            b.append( "md", md.toBSON() );
-            obj = b.obj();
-        }
+    old = Entry(ident, res.getValue());
+    LOG(1) << "stored meta data for " << ns << " @ " << res.getValue();
+    return Status::OK();
+}
 
-        StatusWith<RecordId> res = _rs->insertRecord( opCtx, obj.objdata(), obj.objsize(), false );
-        if ( !res.isOK() )
-            return res.getStatus();
+std::string KVCatalog::getCollectionIdent(StringData ns) const {
+    stdx::lock_guard<stdx::mutex> lk(_identsLock);
+    NSToIdentMap::const_iterator it = _idents.find(ns.toString());
+    invariant(it != _idents.end());
+    return it->second.ident;
+}
 
-        old = Entry( ident, res.getValue() );
-        LOG(1) << "stored meta data for " << ns << " @ " << res.getValue();
-        return Status::OK();
-    }
+std::string KVCatalog::getIndexIdent(OperationContext* opCtx,
+                                     StringData ns,
+                                     StringData idxName) const {
+    BSONObj obj = _findEntry(opCtx, ns);
+    BSONObj idxIdent = obj["idxIdent"].Obj();
+    return idxIdent[idxName].String();
+}
 
-    std::string KVCatalog::getCollectionIdent( StringData ns ) const {
-        stdx::lock_guard<stdx::mutex> lk( _identsLock );
-        NSToIdentMap::const_iterator it = _idents.find( ns.toString() );
-        invariant( it != _idents.end() );
-        return it->second.ident;
+BSONObj KVCatalog::_findEntry(OperationContext* opCtx, StringData ns, RecordId* out) const {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_S));
     }
 
-    std::string KVCatalog::getIndexIdent( OperationContext* opCtx,
-                                          StringData ns,
-                                          StringData idxName ) const {
-        BSONObj obj = _findEntry( opCtx, ns );
-        BSONObj idxIdent = obj["idxIdent"].Obj();
-        return idxIdent[idxName].String();
+    RecordId dl;
+    {
+        stdx::lock_guard<stdx::mutex> lk(_identsLock);
+        NSToIdentMap::const_iterator it = _idents.find(ns.toString());
+        invariant(it != _idents.end());
+        dl = it->second.storedLoc;
     }
 
-    BSONObj KVCatalog::_findEntry( OperationContext* opCtx,
-                                   StringData ns,
-                                   RecordId* out ) const {
-
-        std::unique_ptr<Lock::ResourceLock> rLk;
-        if (!_isRsThreadSafe && opCtx->lockState()) {
-            rLk.reset(new Lock::ResourceLock(opCtx->lockState(),
-                                             resourceIdCatalogMetadata,
-                                             MODE_S));
-        }
+    LOG(1) << "looking up metadata for: " << ns << " @ " << dl;
+    RecordData data;
+    if (!_rs->findRecord(opCtx, dl, &data)) {
+        // since the in memory meta data isn't managed with mvcc
+        // its possible for different transactions to see slightly
+        // different things, which is ok via the locking above.
+        return BSONObj();
+    }
 
-        RecordId dl;
-        {
-            stdx::lock_guard<stdx::mutex> lk( _identsLock );
-            NSToIdentMap::const_iterator it = _idents.find( ns.toString() );
-            invariant( it != _idents.end() );
-            dl = it->second.storedLoc;
-        }
+    if (out)
+        *out = dl;
 
-        LOG(1) << "looking up metadata for: " << ns << " @ " << dl;
-        RecordData data;
-        if ( !_rs->findRecord( opCtx, dl, &data ) ) {
-            // since the in memory meta data isn't managed with mvcc
-            // its possible for different transactions to see slightly
-            // different things, which is ok via the locking above.
-            return BSONObj();
-        }
-
-        if (out)
-            *out = dl;
+    return data.releaseToBson().getOwned();
+}
 
-        return data.releaseToBson().getOwned();
+const BSONCollectionCatalogEntry::MetaData KVCatalog::getMetaData(OperationContext* opCtx,
+                                                                  StringData ns) {
+    BSONObj obj = _findEntry(opCtx, ns);
+    LOG(3) << " fetched CCE metadata: " << obj;
+    BSONCollectionCatalogEntry::MetaData md;
+    const BSONElement mdElement = obj["md"];
+    if (mdElement.isABSONObj()) {
+        LOG(3) << "returning metadata: " << mdElement;
+        md.parse(mdElement.Obj());
     }
+    return md;
+}
 
-    const BSONCollectionCatalogEntry::MetaData KVCatalog::getMetaData( OperationContext* opCtx,
-                                                                       StringData ns ) {
-        BSONObj obj = _findEntry( opCtx, ns );
-        LOG(3) << " fetched CCE metadata: " << obj;
-        BSONCollectionCatalogEntry::MetaData md;
-        const BSONElement mdElement = obj["md"];
-        if ( mdElement.isABSONObj() ) {
-            LOG(3) << "returning metadata: " << mdElement;
-            md.parse( mdElement.Obj() );
-        }
-        return md;
+void KVCatalog::putMetaData(OperationContext* opCtx,
+                            StringData ns,
+                            BSONCollectionCatalogEntry::MetaData& md) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
     }
 
-    void KVCatalog::putMetaData( OperationContext* opCtx,
-                                 StringData ns,
-                                 BSONCollectionCatalogEntry::MetaData& md ) {
-
-        std::unique_ptr<Lock::ResourceLock> rLk;
-        if (!_isRsThreadSafe && opCtx->lockState()) {
-            rLk.reset(new Lock::ResourceLock(opCtx->lockState(),
-                                             resourceIdCatalogMetadata,
-                                             MODE_X));
+    RecordId loc;
+    BSONObj obj = _findEntry(opCtx, ns, &loc);
+
+    {
+        // rebuilt doc
+        BSONObjBuilder b;
+        b.append("md", md.toBSON());
+
+        BSONObjBuilder newIdentMap;
+        BSONObj oldIdentMap;
+        if (obj["idxIdent"].isABSONObj())
+            oldIdentMap = obj["idxIdent"].Obj();
+
+        // fix ident map
+        for (size_t i = 0; i < md.indexes.size(); i++) {
+            string name = md.indexes[i].name();
+            BSONElement e = oldIdentMap[name];
+            if (e.type() == String) {
+                newIdentMap.append(e);
+                continue;
+            }
+            // missing, create new
+            newIdentMap.append(name, _newUniqueIdent(ns, "index"));
         }
+        b.append("idxIdent", newIdentMap.obj());
 
-        RecordId loc;
-        BSONObj obj = _findEntry( opCtx, ns, &loc );
-
-        {
-            // rebuilt doc
-            BSONObjBuilder b;
-            b.append( "md", md.toBSON() );
-
-            BSONObjBuilder newIdentMap;
-            BSONObj oldIdentMap;
-            if ( obj["idxIdent"].isABSONObj() )
-                oldIdentMap = obj["idxIdent"].Obj();
-
-            // fix ident map
-            for ( size_t i = 0; i < md.indexes.size(); i++ ) {
-                string name = md.indexes[i].name();
-                BSONElement e = oldIdentMap[name];
-                if ( e.type() == String ) {
-                    newIdentMap.append( e );
-                    continue;
-                }
-                // missing, create new
-                newIdentMap.append( name, _newUniqueIdent(ns, "index") );
-            }
-            b.append( "idxIdent", newIdentMap.obj() );
+        // add whatever is left
+        b.appendElementsUnique(obj);
+        obj = b.obj();
+    }
 
-            // add whatever is left
-            b.appendElementsUnique( obj );
-            obj = b.obj();
-        }
+    LOG(3) << "recording new metadata: " << obj;
+    StatusWith<RecordId> status =
+        _rs->updateRecord(opCtx, loc, obj.objdata(), obj.objsize(), false, NULL);
+    fassert(28521, status.getStatus());
+    invariant(status.getValue() == loc);
+}
 
-        LOG(3) << "recording new metadata: " << obj;
-        StatusWith<RecordId> status = _rs->updateRecord( opCtx,
-                                                        loc,
-                                                        obj.objdata(),
-                                                        obj.objsize(),
-                                                        false,
-                                                        NULL );
-        fassert( 28521, status.getStatus() );
-        invariant( status.getValue() == loc );
+Status KVCatalog::renameCollection(OperationContext* opCtx,
+                                   StringData fromNS,
+                                   StringData toNS,
+                                   bool stayTemp) {
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
     }
 
-    Status KVCatalog::renameCollection( OperationContext* opCtx,
-                                        StringData fromNS,
-                                        StringData toNS,
-                                        bool stayTemp ) {
+    RecordId loc;
+    BSONObj old = _findEntry(opCtx, fromNS, &loc).getOwned();
+    {
+        BSONObjBuilder b;
 
-        std::unique_ptr<Lock::ResourceLock> rLk;
-        if (!_isRsThreadSafe && opCtx->lockState()) {
-            rLk.reset(new Lock::ResourceLock(opCtx->lockState(),
-                                             resourceIdCatalogMetadata,
-                                             MODE_X));
-        }
+        b.append("ns", toNS);
 
-        RecordId loc;
-        BSONObj old = _findEntry( opCtx, fromNS, &loc ).getOwned();
-        {
-            BSONObjBuilder b;
-
-            b.append( "ns", toNS );
-
-            BSONCollectionCatalogEntry::MetaData md;
-            md.parse( old["md"].Obj() );
-            md.rename( toNS );
-            if ( !stayTemp )
-                md.options.temp = false;
-            b.append( "md", md.toBSON() );
-
-            b.appendElementsUnique( old );
-
-            BSONObj obj = b.obj();
-            StatusWith<RecordId> status = _rs->updateRecord( opCtx,
-                                                            loc,
-                                                            obj.objdata(),
-                                                            obj.objsize(),
-                                                            false,
-                                                            NULL );
-            fassert( 28522, status.getStatus() );
-            invariant( status.getValue() == loc );
-        }
+        BSONCollectionCatalogEntry::MetaData md;
+        md.parse(old["md"].Obj());
+        md.rename(toNS);
+        if (!stayTemp)
+            md.options.temp = false;
+        b.append("md", md.toBSON());
+
+        b.appendElementsUnique(old);
+
+        BSONObj obj = b.obj();
+        StatusWith<RecordId> status =
+            _rs->updateRecord(opCtx, loc, obj.objdata(), obj.objsize(), false, NULL);
+        fassert(28522, status.getStatus());
+        invariant(status.getValue() == loc);
+    }
 
-        stdx::lock_guard<stdx::mutex> lk( _identsLock );
-        const NSToIdentMap::iterator fromIt = _idents.find(fromNS.toString());
-        invariant(fromIt != _idents.end());
+    stdx::lock_guard<stdx::mutex> lk(_identsLock);
+    const NSToIdentMap::iterator fromIt = _idents.find(fromNS.toString());
+    invariant(fromIt != _idents.end());
 
-        opCtx->recoveryUnit()->registerChange(new RemoveIdentChange(this, fromNS, fromIt->second));
-        opCtx->recoveryUnit()->registerChange(new AddIdentChange(this, toNS));
+    opCtx->recoveryUnit()->registerChange(new RemoveIdentChange(this, fromNS, fromIt->second));
+    opCtx->recoveryUnit()->registerChange(new AddIdentChange(this, toNS));
 
-        _idents.erase(fromIt);
-        _idents[toNS.toString()] = Entry( old["ident"].String(), loc );
+    _idents.erase(fromIt);
+    _idents[toNS.toString()] = Entry(old["ident"].String(), loc);
 
-        return Status::OK();
-    }
+    return Status::OK();
+}
 
-    Status KVCatalog::dropCollection( OperationContext* opCtx,
-                                      StringData ns ) {
-        invariant( opCtx->lockState() == NULL ||
-                   opCtx->lockState()->isDbLockedForMode( nsToDatabaseSubstring(ns), MODE_X ) );
-        std::unique_ptr<Lock::ResourceLock> rLk;
-        if (!_isRsThreadSafe && opCtx->lockState()) {
-            rLk.reset(new Lock::ResourceLock(opCtx->lockState(),
-                                             resourceIdCatalogMetadata,
-                                             MODE_X));
-        }
+Status KVCatalog::dropCollection(OperationContext* opCtx, StringData ns) {
+    invariant(opCtx->lockState() == NULL ||
+              opCtx->lockState()->isDbLockedForMode(nsToDatabaseSubstring(ns), MODE_X));
+    std::unique_ptr<Lock::ResourceLock> rLk;
+    if (!_isRsThreadSafe && opCtx->lockState()) {
+        rLk.reset(new Lock::ResourceLock(opCtx->lockState(), resourceIdCatalogMetadata, MODE_X));
+    }
 
-        stdx::lock_guard<stdx::mutex> lk( _identsLock );
-        const NSToIdentMap::iterator it = _idents.find(ns.toString());
-        if (it == _idents.end()) {
-            return Status( ErrorCodes::NamespaceNotFound, "collection not found" );
-        }
+    stdx::lock_guard<stdx::mutex> lk(_identsLock);
+    const NSToIdentMap::iterator it = _idents.find(ns.toString());
+    if (it == _idents.end()) {
+        return Status(ErrorCodes::NamespaceNotFound, "collection not found");
+    }
 
-        opCtx->recoveryUnit()->registerChange(new RemoveIdentChange(this, ns, it->second));
+    opCtx->recoveryUnit()->registerChange(new RemoveIdentChange(this, ns, it->second));
 
-        LOG(1) << "deleting metadata for " << ns << " @ " << it->second.storedLoc;
-        _rs->deleteRecord( opCtx, it->second.storedLoc );
-        _idents.erase(it);
+    LOG(1) << "deleting metadata for " << ns << " @ " << it->second.storedLoc;
+    _rs->deleteRecord(opCtx, it->second.storedLoc);
+    _idents.erase(it);
 
-        return Status::OK();
-    }
+    return Status::OK();
+}
 
-    std::vector<std::string> KVCatalog::getAllIdentsForDB( StringData db ) const {
-        std::vector<std::string> v;
+std::vector<std::string> KVCatalog::getAllIdentsForDB(StringData db) const {
+    std::vector<std::string> v;
 
-        {
-            stdx::lock_guard<stdx::mutex> lk( _identsLock );
-            for ( NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it ) {
-                NamespaceString ns( it->first );
-                if ( ns.db() != db )
-                    continue;
-                v.push_back( it->second.ident );
-            }
+    {
+        stdx::lock_guard<stdx::mutex> lk(_identsLock);
+        for (NSToIdentMap::const_iterator it = _idents.begin(); it != _idents.end(); ++it) {
+            NamespaceString ns(it->first);
+            if (ns.db() != db)
+                continue;
+            v.push_back(it->second.ident);
         }
-
-        return v;
     }
 
-    std::vector<std::string> KVCatalog::getAllIdents( OperationContext* opCtx ) const {
-        std::vector<std::string> v;
+    return v;
+}
 
-        auto cursor = _rs->getCursor(opCtx);
-        while (auto record = cursor->next()) {
-            BSONObj obj = record->data.releaseToBson();
-            v.push_back( obj["ident"].String() );
+std::vector<std::string> KVCatalog::getAllIdents(OperationContext* opCtx) const {
+    std::vector<std::string> v;
 
-            BSONElement e = obj["idxIdent"];
-            if ( !e.isABSONObj() )
-                continue;
-            BSONObj idxIdent = e.Obj();
+    auto cursor = _rs->getCursor(opCtx);
+    while (auto record = cursor->next()) {
+        BSONObj obj = record->data.releaseToBson();
+        v.push_back(obj["ident"].String());
 
-            BSONObjIterator sub( idxIdent );
-            while ( sub.more() ) {
-                BSONElement e = sub.next();
-                v.push_back( e.String() );
-            }
-        }
+        BSONElement e = obj["idxIdent"];
+        if (!e.isABSONObj())
+            continue;
+        BSONObj idxIdent = e.Obj();
 
-        return v;
+        BSONObjIterator sub(idxIdent);
+        while (sub.more()) {
+            BSONElement e = sub.next();
+            v.push_back(e.String());
+        }
     }
 
-    bool KVCatalog::isUserDataIdent( StringData ident ) const {
-        return
-            ident.find( "index-" ) != std::string::npos ||
-            ident.find( "index/" ) != std::string::npos ||
-            ident.find( "collection-" ) != std::string::npos ||
-            ident.find( "collection/" ) != std::string::npos;
-    }
+    return v;
+}
 
+bool KVCatalog::isUserDataIdent(StringData ident) const {
+    return ident.find("index-") != std::string::npos || ident.find("index/") != std::string::npos ||
+        ident.find("collection-") != std::string::npos ||
+        ident.find("collection/") != std::string::npos;
+}
 }