'Rollback target system to consistent state.
The strategy is to find the latest timestamp in the target system and
the largest timestamp in the oplog less than the latest target system
timestamp. This defines the rollback window and we just roll these
back until the oplog and target system are in consistent states.'

def rollback(self):

    LOG.debug('OplogThread:    Initiating    rollback    sequence    to    bring    system    into    a    consistent    state.')
    last_docs = []
    for dm in self.doc_managers:
        dm.commit()
        last_docs.append(dm.get_last_doc())
    last_inserted_doc = max(last_docs, key=(lambda x: (x['_ts'] if x else float('-inf'))))
    if (last_inserted_doc is None):
        return None
    target_ts = util.long_to_bson_ts(last_inserted_doc['_ts'])
    last_oplog_entry = util.retry_until_ok(self.oplog.find_one, {'ts': {'$lte': target_ts}, 'op': {'$ne': 'n'}}, sort=[('$natural', pymongo.DESCENDING)])
    LOG.debug(('OplogThread:    last    oplog    entry    is    %s' % str(last_oplog_entry)))
    if (last_oplog_entry is None):
        return None
    rollback_cutoff_ts = last_oplog_entry['ts']
    start_ts = util.bson_ts_to_long(rollback_cutoff_ts)
    end_ts = last_inserted_doc['_ts']
    for dm in self.doc_managers:
        rollback_set = {}
        for doc in dm.search(start_ts, end_ts):
            if (doc['ns'] in rollback_set):
                rollback_set[doc['ns']].append(doc)
            else:
                rollback_set[doc['ns']] = [doc]
        for (namespace, doc_list) in rollback_set.items():
            original_namespace = self.namespace_config.unmap_namespace(namespace)
            if (not original_namespace):
                original_namespace = namespace
            (database, coll) = original_namespace.split('.', 1)
            obj_id = bson.objectid.ObjectId
            bson_obj_id_list = [obj_id(doc['_id']) for doc in doc_list]
            client = (self.mongos_client or self.primary_client)
            to_update = util.retry_until_ok(client[database][coll].find, {'_id': {'$in': bson_obj_id_list}}, projection=self.namespace_config.projection(original_namespace))
            doc_hash = {}
            for doc in doc_list:
                doc_hash[bson.objectid.ObjectId(doc['_id'])] = doc
            to_index = []
            def collect_existing_docs():
                for doc in to_update:
                    if (doc['_id'] in doc_hash):
                        del doc_hash[doc['_id']]
                        to_index.append(doc)
            retry_until_ok(collect_existing_docs)
            LOG.debug('OplogThread:    Rollback,    removing    inconsistent    docs.')
            remov_inc = 0
            for document_id in doc_hash:
                try:
                    dm.remove(document_id, namespace, util.bson_ts_to_long(rollback_cutoff_ts))
                    remov_inc += 1
                    LOG.debug(('OplogThread:    Rollback,    removed    %r    ' % doc))
                except errors.OperationFailed:
                    LOG.warning(('Could    not    delete    document    during    rollback:    %r    This    can    happen    if    this    document    was    already    removed    by    another    rollback    happening    at    the    same    time.' % doc))
            LOG.debug(('OplogThread:    Rollback,    removed    %d    docs.' % remov_inc))
            LOG.debug('OplogThread:    Rollback,    inserting    documents    from    mongo.')
            insert_inc = 0
            fail_insert_inc = 0
            for doc in to_index:
                try:
                    insert_inc += 1
                    dm.upsert(doc, namespace, util.bson_ts_to_long(rollback_cutoff_ts))
                except errors.OperationFailed:
                    fail_insert_inc += 1
                    LOG.exception(('OplogThread:    Rollback,    Unable    to    insert    %r' % doc))
    LOG.debug(('OplogThread:    Rollback,    Successfully    inserted    %d        documents    and    failed    to    insert    %d    documents.        Returning    a    rollback    cutoff    time    of    %s    ' % (insert_inc, fail_insert_inc, str(rollback_cutoff_ts))))
    return rollback_cutoff_ts
