From baeae199516009e14584dbdff16f82f096c8afb9 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Wed, 25 Jan 2017 19:05:48 +0100
Subject: [PATCH 01/19] Add experimental run interface

---
 bigchaindb/backend/mongodb/connection.py |   8 +
 bigchaindb/backend/mongodb/query.py      | 235 +++++++++++++----------
 bigchaindb/utils.py                      |  28 +++
 tests/test_utils.py                      |   8 +
 4 files changed, 180 insertions(+), 99 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 19731161..eaa9852e 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -5,6 +5,7 @@ from pymongo import MongoClient
 from pymongo import errors
 
 import bigchaindb
+from bigchaindb.utils import Lazy
 from bigchaindb.common import exceptions
 from bigchaindb.backend.connection import Connection
 
@@ -43,6 +44,9 @@ class MongoDBConnection(Connection):
     def db(self):
         return self.conn[self.dbname]
 
+    def run(self, query):
+        return query.run(self.db)
+
     def _connect(self):
         # we should only return a connection if the replica set is
         # initialized. initialize_replica_set will check if the
@@ -60,6 +64,10 @@ class MongoDBConnection(Connection):
                     time.sleep(2**i)
 
 
+def table(name):
+    return Lazy()[name]
+
+
 def initialize_replica_set():
     """Initialize a replica set. If already initialized skip."""
 
diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py
index c4e3cdc8..ee7562c1 100644
--- a/bigchaindb/backend/mongodb/query.py
+++ b/bigchaindb/backend/mongodb/query.py
@@ -5,10 +5,11 @@ from time import time
 from pymongo import ReturnDocument
 from pymongo import errors
 
+
 from bigchaindb import backend
 from bigchaindb.common.exceptions import CyclicBlockchainError
 from bigchaindb.backend.utils import module_dispatch_registrar
-from bigchaindb.backend.mongodb.connection import MongoDBConnection
+from bigchaindb.backend.mongodb.connection import MongoDBConnection, table
 
 
 register_query = module_dispatch_registrar(backend.query)
@@ -17,7 +18,9 @@ register_query = module_dispatch_registrar(backend.query)
 @register_query(MongoDBConnection)
 def write_transaction(conn, signed_transaction):
     try:
-        return conn.db['backlog'].insert_one(signed_transaction)
+        return conn.run(
+            table('backlog')
+            .insert_one(signed_transaction))
     except errors.DuplicateKeyError:
         return
 
@@ -26,40 +29,49 @@ def write_transaction(conn, signed_transaction):
 def update_transaction(conn, transaction_id, doc):
     # with mongodb we need to add update operators to the doc
     doc = {'$set': doc}
-    return conn.db['backlog']\
-               .find_one_and_update({'id': transaction_id},
-                                    doc,
-                                    return_document=ReturnDocument.AFTER)
+    return conn.run(
+        table('backlog')
+        .find_one_and_update(
+            {'id': transaction_id},
+            doc,
+            return_document=ReturnDocument.AFTER))
 
 
 @register_query(MongoDBConnection)
 def delete_transaction(conn, *transaction_id):
-    return conn.db['backlog'].delete_many({'id': {'$in': transaction_id}})
+    return conn.run(
+        table('backlog')
+        .delete_many({'id': {'$in': transaction_id}}))
 
 
 @register_query(MongoDBConnection)
 def get_stale_transactions(conn, reassign_delay):
-    return conn.db['backlog']\
-            .find({'assignment_timestamp': {'$lt': time() - reassign_delay}},
-                  projection={'_id': False})
+    return conn.run(
+        table('backlog')
+        .find({'assignment_timestamp': {'$lt': time() - reassign_delay}},
+              projection={'_id': False}))
 
 
 @register_query(MongoDBConnection)
 def get_transaction_from_block(conn, transaction_id, block_id):
     try:
-        return conn.db['bigchain'].aggregate([
-            {'$match': {'id': block_id}},
-            {'$project': {
-                'block.transactions': {
-                    '$filter': {
-                        'input': '$block.transactions',
-                        'as': 'transaction',
-                        'cond': {
-                            '$eq': ['$$transaction.id', transaction_id]
+        return conn.run(
+            table('bigchain')
+            .aggregate([
+                {'$match': {'id': block_id}},
+                {'$project': {
+                    'block.transactions': {
+                        '$filter': {
+                            'input': '$block.transactions',
+                            'as': 'transaction',
+                            'cond': {
+                                '$eq': ['$$transaction.id', transaction_id]
+                                }
+                            }
                         }
-                    }
-                }
-            }}]).next()['block']['transactions'].pop()
+                    }}])
+            .next()['block']['transactions']
+            .pop())
     except (StopIteration, IndexError):
         # StopIteration is raised if the block was not found
         # IndexError is returned if the block is found but no transactions
@@ -69,33 +81,38 @@ def get_transaction_from_block(conn, transaction_id, block_id):
 
 @register_query(MongoDBConnection)
 def get_transaction_from_backlog(conn, transaction_id):
-    return conn.db['backlog']\
-               .find_one({'id': transaction_id},
-                         projection={'_id': False, 'assignee': False,
-                                     'assignment_timestamp': False})
+    return conn.run(
+        table('backlog')
+        .find_one({'id': transaction_id},
+                  projection={'_id': False,
+                              'assignee': False,
+                              'assignment_timestamp': False}))
 
 
 @register_query(MongoDBConnection)
 def get_blocks_status_from_transaction(conn, transaction_id):
-    return conn.db['bigchain']\
-            .find({'block.transactions.id': transaction_id},
-                  projection=['id', 'block.voters'])
+    return conn.run(
+        table('bigchain')
+        .find({'block.transactions.id': transaction_id},
+              projection=['id', 'block.voters']))
 
 
 @register_query(MongoDBConnection)
 def get_asset_by_id(conn, asset_id):
-    cursor = conn.db['bigchain'].aggregate([
-        {'$match': {
-            'block.transactions.id': asset_id,
-            'block.transactions.operation': 'CREATE'
-        }},
-        {'$unwind': '$block.transactions'},
-        {'$match': {
-            'block.transactions.id': asset_id,
-            'block.transactions.operation': 'CREATE'
-        }},
-        {'$project': {'block.transactions.asset': True}}
-    ])
+    cursor = conn.run(
+        table('bigchain')
+        .aggregate([
+            {'$match': {
+                'block.transactions.id': asset_id,
+                'block.transactions.operation': 'CREATE'
+            }},
+            {'$unwind': '$block.transactions'},
+            {'$match': {
+                'block.transactions.id': asset_id,
+                'block.transactions.operation': 'CREATE'
+            }},
+            {'$project': {'block.transactions.asset': True}}
+        ]))
     # we need to access some nested fields before returning so lets use a
     # generator to avoid having to read all records on the cursor at this point
     return (elem['block']['transactions'] for elem in cursor)
@@ -103,13 +120,14 @@ def get_asset_by_id(conn, asset_id):
 
 @register_query(MongoDBConnection)
 def get_spent(conn, transaction_id, output):
-    cursor = conn.db['bigchain'].aggregate([
-        {'$unwind': '$block.transactions'},
-        {'$match': {
-            'block.transactions.inputs.fulfills.txid': transaction_id,
-            'block.transactions.inputs.fulfills.output': output
-        }}
-    ])
+    cursor = conn.run(
+        table('bigchain').aggregate([
+            {'$unwind': '$block.transactions'},
+            {'$match': {
+                'block.transactions.inputs.fulfills.txid': transaction_id,
+                'block.transactions.inputs.fulfills.output': output
+            }}
+        ]))
     # we need to access some nested fields before returning so lets use a
     # generator to avoid having to read all records on the cursor at this point
     return (elem['block']['transactions'] for elem in cursor)
@@ -117,14 +135,16 @@ def get_spent(conn, transaction_id, output):
 
 @register_query(MongoDBConnection)
 def get_owned_ids(conn, owner):
-    cursor = conn.db['bigchain'].aggregate([
-        {'$unwind': '$block.transactions'},
-        {'$match': {
-            'block.transactions.outputs.public_keys': {
-                '$elemMatch': {'$eq': owner}
-            }
-        }}
-    ])
+    cursor = conn.run(
+        table('bigchain')
+        .aggregate([
+            {'$unwind': '$block.transactions'},
+            {'$match': {
+                'block.transactions.outputs.public_keys': {
+                    '$elemMatch': {'$eq': owner}
+                }
+            }}
+        ]))
     # we need to access some nested fields before returning so lets use a
     # generator to avoid having to read all records on the cursor at this point
     return (elem['block']['transactions'] for elem in cursor)
@@ -132,66 +152,80 @@ def get_owned_ids(conn, owner):
 
 @register_query(MongoDBConnection)
 def get_votes_by_block_id(conn, block_id):
-    return conn.db['votes']\
-            .find({'vote.voting_for_block': block_id},
-                  projection={'_id': False})
+    return conn.run(
+        table('votes')
+        .find({'vote.voting_for_block': block_id},
+              projection={'_id': False}))
 
 
 @register_query(MongoDBConnection)
 def get_votes_by_block_id_and_voter(conn, block_id, node_pubkey):
-    return conn.db['votes']\
-            .find({'vote.voting_for_block': block_id,
-                   'node_pubkey': node_pubkey},
-                  projection={'_id': False})
+    return conn.run(
+        table('votes')
+        .find({'vote.voting_for_block': block_id,
+               'node_pubkey': node_pubkey},
+              projection={'_id': False}))
 
 
 @register_query(MongoDBConnection)
 def write_block(conn, block):
-    return conn.db['bigchain'].insert_one(block.to_dict())
+    return conn.run(
+        table('bigchain')
+        .insert_one(block.to_dict()))
 
 
 @register_query(MongoDBConnection)
 def get_block(conn, block_id):
-    return conn.db['bigchain'].find_one({'id': block_id},
-                                        projection={'_id': False})
+    return conn.run(
+        table('bigchain')
+        .find_one({'id': block_id},
+                  projection={'_id': False}))
 
 
 @register_query(MongoDBConnection)
 def has_transaction(conn, transaction_id):
-    return bool(conn.db['bigchain']
-                .find_one({'block.transactions.id': transaction_id}))
+    return bool(conn.run(
+        table('bigchain')
+        .find_one({'block.transactions.id': transaction_id})))
 
 
 @register_query(MongoDBConnection)
 def count_blocks(conn):
-    return conn.db['bigchain'].count()
+    return conn.run(
+        table('bigchain')
+        .count())
 
 
 @register_query(MongoDBConnection)
 def count_backlog(conn):
-    return conn.db['backlog'].count()
+    return conn.run(
+        table('backlog')
+        .count())
 
 
 @register_query(MongoDBConnection)
 def write_vote(conn, vote):
-    conn.db['votes'].insert_one(vote)
+    conn.run(table('votes').insert_one(vote))
     vote.pop('_id')
     return vote
 
 
 @register_query(MongoDBConnection)
 def get_genesis_block(conn):
-    return conn.db['bigchain'].find_one(
-        {'block.transactions.0.operation': 'GENESIS'},
-        {'_id': False}
-    )
+    return conn.run(
+        table('bigchain')
+            .find_one(
+            {'block.transactions.0.operation': 'GENESIS'},
+            {'_id': False}
+        ))
 
 
 @register_query(MongoDBConnection)
 def get_last_voted_block(conn, node_pubkey):
-    last_voted = conn.db['votes']\
-                  .find({'node_pubkey': node_pubkey},
-                        sort=[('vote.timestamp', -1)])
+    last_voted = conn.run(
+            table('votes')
+            .find({'node_pubkey': node_pubkey},
+                  sort=[('vote.timestamp', -1)]))
 
     # pymongo seems to return a cursor even if there are no results
     # so we actually need to check the count
@@ -219,21 +253,22 @@ def get_last_voted_block(conn, node_pubkey):
 
 @register_query(MongoDBConnection)
 def get_unvoted_blocks(conn, node_pubkey):
-    return conn.db['bigchain'].aggregate([
-        {'$lookup': {
-            'from': 'votes',
-            'localField': 'id',
-            'foreignField': 'vote.voting_for_block',
-            'as': 'votes'
-        }},
-        {'$match': {
-            'votes.node_pubkey': {'$ne': node_pubkey},
-            'block.transactions.operation': {'$ne': 'GENESIS'}
-        }},
-        {'$project': {
-            'votes': False, '_id': False
-        }}
-    ])
+    return conn.run(
+        table('bigchain').aggregate([
+            {'$lookup': {
+                'from': 'votes',
+                'localField': 'id',
+                'foreignField': 'vote.voting_for_block',
+                'as': 'votes'
+            }},
+            {'$match': {
+                'votes.node_pubkey': {'$ne': node_pubkey},
+                'block.transactions.operation': {'$ne': 'GENESIS'}
+            }},
+            {'$project': {
+                'votes': False, '_id': False
+            }}
+        ]))
 
 
 @register_query(MongoDBConnection)
@@ -243,10 +278,12 @@ def get_txids_filtered(conn, asset_id, operation=None):
     if operation:
         match['block.transactions.operation'] = operation
 
-    cursor = conn.db['bigchain'].aggregate([
-        {'$match': match},
-        {'$unwind': '$block.transactions'},
-        {'$match': match},
-        {'$project': {'block.transactions.id': True}}
-    ])
+    cursor = conn.run(
+        table('bigchain')
+        .aggregate([
+            {'$match': match},
+            {'$unwind': '$block.transactions'},
+            {'$match': match},
+            {'$project': {'block.transactions.id': True}}
+        ]))
     return (r['block']['transactions']['id'] for r in cursor)
diff --git a/bigchaindb/utils.py b/bigchaindb/utils.py
index 434c6376..e89c3b6f 100644
--- a/bigchaindb/utils.py
+++ b/bigchaindb/utils.py
@@ -157,3 +157,31 @@ def is_genesis_block(block):
         return block.transactions[0].operation == 'GENESIS'
     except AttributeError:
         return block['block']['transactions'][0]['operation'] == 'GENESIS'
+
+
+class Lazy:
+
+    def __init__(self):
+        self.stack = []
+
+    def __getattr__(self, name):
+        self.stack.append(name)
+        return self
+
+    def __call__(self, *args, **kwargs):
+        self.stack.append((args, kwargs))
+        return self
+
+    def __getitem__(self, key):
+        self.stack.append('__getitem__')
+        self.stack.append(([key], {}))
+        return self
+
+    def run(self, instance):
+        last = instance
+
+        for method, (args, kwargs) in zip(self.stack[::2], self.stack[1::2]):
+            last = getattr(last, method)(*args, **kwargs)
+
+        self.stack = []
+        return last
diff --git a/tests/test_utils.py b/tests/test_utils.py
index f76ba6d9..47343ace 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -137,3 +137,11 @@ def test_is_genesis_block_returns_true_if_genesis(b):
     from bigchaindb.utils import is_genesis_block
     genesis_block = b.prepare_genesis_block()
     assert is_genesis_block(genesis_block)
+
+
+def test_lazy_execution():
+    from bigchaindb.utils import Lazy
+    l = Lazy()
+    l.split(',')[1].split(' ').pop(1).strip()
+    result = l.run('Like humans, cats tend to favor one paw over another')
+    assert result == 'cats'

From 83afab4958bcfbebd84818376467246ac277b69a Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 26 Jan 2017 10:22:43 +0100
Subject: [PATCH 02/19] Fix indentation problem

---
 bigchaindb/backend/mongodb/query.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py
index ee7562c1..f1acf601 100644
--- a/bigchaindb/backend/mongodb/query.py
+++ b/bigchaindb/backend/mongodb/query.py
@@ -214,7 +214,7 @@ def write_vote(conn, vote):
 def get_genesis_block(conn):
     return conn.run(
         table('bigchain')
-            .find_one(
+        .find_one(
             {'block.transactions.0.operation': 'GENESIS'},
             {'_id': False}
         ))

From ca49718d7e9f14ae997c4e00a66098c9b19581cb Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 26 Jan 2017 11:23:50 +0100
Subject: [PATCH 03/19] s/table/collection/g

---
 bigchaindb/backend/mongodb/connection.py |  2 +-
 bigchaindb/backend/mongodb/query.py      | 46 ++++++++++++------------
 2 files changed, 24 insertions(+), 24 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index eaa9852e..61af4469 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -64,7 +64,7 @@ class MongoDBConnection(Connection):
                     time.sleep(2**i)
 
 
-def table(name):
+def collection(name):
     return Lazy()[name]
 
 
diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py
index f1acf601..14c1ae21 100644
--- a/bigchaindb/backend/mongodb/query.py
+++ b/bigchaindb/backend/mongodb/query.py
@@ -9,7 +9,7 @@ from pymongo import errors
 from bigchaindb import backend
 from bigchaindb.common.exceptions import CyclicBlockchainError
 from bigchaindb.backend.utils import module_dispatch_registrar
-from bigchaindb.backend.mongodb.connection import MongoDBConnection, table
+from bigchaindb.backend.mongodb.connection import MongoDBConnection, collection
 
 
 register_query = module_dispatch_registrar(backend.query)
@@ -19,7 +19,7 @@ register_query = module_dispatch_registrar(backend.query)
 def write_transaction(conn, signed_transaction):
     try:
         return conn.run(
-            table('backlog')
+            collection('backlog')
             .insert_one(signed_transaction))
     except errors.DuplicateKeyError:
         return
@@ -30,7 +30,7 @@ def update_transaction(conn, transaction_id, doc):
     # with mongodb we need to add update operators to the doc
     doc = {'$set': doc}
     return conn.run(
-        table('backlog')
+        collection('backlog')
         .find_one_and_update(
             {'id': transaction_id},
             doc,
@@ -40,14 +40,14 @@ def update_transaction(conn, transaction_id, doc):
 @register_query(MongoDBConnection)
 def delete_transaction(conn, *transaction_id):
     return conn.run(
-        table('backlog')
+        collection('backlog')
         .delete_many({'id': {'$in': transaction_id}}))
 
 
 @register_query(MongoDBConnection)
 def get_stale_transactions(conn, reassign_delay):
     return conn.run(
-        table('backlog')
+        collection('backlog')
         .find({'assignment_timestamp': {'$lt': time() - reassign_delay}},
               projection={'_id': False}))
 
@@ -56,7 +56,7 @@ def get_stale_transactions(conn, reassign_delay):
 def get_transaction_from_block(conn, transaction_id, block_id):
     try:
         return conn.run(
-            table('bigchain')
+            collection('bigchain')
             .aggregate([
                 {'$match': {'id': block_id}},
                 {'$project': {
@@ -82,7 +82,7 @@ def get_transaction_from_block(conn, transaction_id, block_id):
 @register_query(MongoDBConnection)
 def get_transaction_from_backlog(conn, transaction_id):
     return conn.run(
-        table('backlog')
+        collection('backlog')
         .find_one({'id': transaction_id},
                   projection={'_id': False,
                               'assignee': False,
@@ -92,7 +92,7 @@ def get_transaction_from_backlog(conn, transaction_id):
 @register_query(MongoDBConnection)
 def get_blocks_status_from_transaction(conn, transaction_id):
     return conn.run(
-        table('bigchain')
+        collection('bigchain')
         .find({'block.transactions.id': transaction_id},
               projection=['id', 'block.voters']))
 
@@ -100,7 +100,7 @@ def get_blocks_status_from_transaction(conn, transaction_id):
 @register_query(MongoDBConnection)
 def get_asset_by_id(conn, asset_id):
     cursor = conn.run(
-        table('bigchain')
+        collection('bigchain')
         .aggregate([
             {'$match': {
                 'block.transactions.id': asset_id,
@@ -121,7 +121,7 @@ def get_asset_by_id(conn, asset_id):
 @register_query(MongoDBConnection)
 def get_spent(conn, transaction_id, output):
     cursor = conn.run(
-        table('bigchain').aggregate([
+        collection('bigchain').aggregate([
             {'$unwind': '$block.transactions'},
             {'$match': {
                 'block.transactions.inputs.fulfills.txid': transaction_id,
@@ -136,7 +136,7 @@ def get_spent(conn, transaction_id, output):
 @register_query(MongoDBConnection)
 def get_owned_ids(conn, owner):
     cursor = conn.run(
-        table('bigchain')
+        collection('bigchain')
         .aggregate([
             {'$unwind': '$block.transactions'},
             {'$match': {
@@ -153,7 +153,7 @@ def get_owned_ids(conn, owner):
 @register_query(MongoDBConnection)
 def get_votes_by_block_id(conn, block_id):
     return conn.run(
-        table('votes')
+        collection('votes')
         .find({'vote.voting_for_block': block_id},
               projection={'_id': False}))
 
@@ -161,7 +161,7 @@ def get_votes_by_block_id(conn, block_id):
 @register_query(MongoDBConnection)
 def get_votes_by_block_id_and_voter(conn, block_id, node_pubkey):
     return conn.run(
-        table('votes')
+        collection('votes')
         .find({'vote.voting_for_block': block_id,
                'node_pubkey': node_pubkey},
               projection={'_id': False}))
@@ -170,14 +170,14 @@ def get_votes_by_block_id_and_voter(conn, block_id, node_pubkey):
 @register_query(MongoDBConnection)
 def write_block(conn, block):
     return conn.run(
-        table('bigchain')
+        collection('bigchain')
         .insert_one(block.to_dict()))
 
 
 @register_query(MongoDBConnection)
 def get_block(conn, block_id):
     return conn.run(
-        table('bigchain')
+        collection('bigchain')
         .find_one({'id': block_id},
                   projection={'_id': False}))
 
@@ -185,27 +185,27 @@ def get_block(conn, block_id):
 @register_query(MongoDBConnection)
 def has_transaction(conn, transaction_id):
     return bool(conn.run(
-        table('bigchain')
+        collection('bigchain')
         .find_one({'block.transactions.id': transaction_id})))
 
 
 @register_query(MongoDBConnection)
 def count_blocks(conn):
     return conn.run(
-        table('bigchain')
+        collection('bigchain')
         .count())
 
 
 @register_query(MongoDBConnection)
 def count_backlog(conn):
     return conn.run(
-        table('backlog')
+        collection('backlog')
         .count())
 
 
 @register_query(MongoDBConnection)
 def write_vote(conn, vote):
-    conn.run(table('votes').insert_one(vote))
+    conn.run(collection('votes').insert_one(vote))
     vote.pop('_id')
     return vote
 
@@ -213,7 +213,7 @@ def write_vote(conn, vote):
 @register_query(MongoDBConnection)
 def get_genesis_block(conn):
     return conn.run(
-        table('bigchain')
+        collection('bigchain')
         .find_one(
             {'block.transactions.0.operation': 'GENESIS'},
             {'_id': False}
@@ -223,7 +223,7 @@ def get_genesis_block(conn):
 @register_query(MongoDBConnection)
 def get_last_voted_block(conn, node_pubkey):
     last_voted = conn.run(
-            table('votes')
+            collection('votes')
             .find({'node_pubkey': node_pubkey},
                   sort=[('vote.timestamp', -1)]))
 
@@ -254,7 +254,7 @@ def get_last_voted_block(conn, node_pubkey):
 @register_query(MongoDBConnection)
 def get_unvoted_blocks(conn, node_pubkey):
     return conn.run(
-        table('bigchain').aggregate([
+        collection('bigchain').aggregate([
             {'$lookup': {
                 'from': 'votes',
                 'localField': 'id',
@@ -279,7 +279,7 @@ def get_txids_filtered(conn, asset_id, operation=None):
         match['block.transactions.operation'] = operation
 
     cursor = conn.run(
-        table('bigchain')
+        collection('bigchain')
         .aggregate([
             {'$match': match},
             {'$unwind': '$block.transactions'},

From 54544f66a050b2333ff61d0b161f9e28dfd0cb61 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 26 Jan 2017 17:12:35 +0100
Subject: [PATCH 04/19] Simplify run function

---
 bigchaindb/utils.py |  7 +++++--
 tests/test_utils.py | 12 ++++++++++++
 2 files changed, 17 insertions(+), 2 deletions(-)

diff --git a/bigchaindb/utils.py b/bigchaindb/utils.py
index e89c3b6f..6ceace6b 100644
--- a/bigchaindb/utils.py
+++ b/bigchaindb/utils.py
@@ -180,8 +180,11 @@ class Lazy:
     def run(self, instance):
         last = instance
 
-        for method, (args, kwargs) in zip(self.stack[::2], self.stack[1::2]):
-            last = getattr(last, method)(*args, **kwargs)
+        for item in self.stack:
+            if isinstance(item, str):
+                last = getattr(last, item)
+            else:
+                last = last(*item[0], **item[1])
 
         self.stack = []
         return last
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 47343ace..fbf5d65d 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -141,7 +141,19 @@ def test_is_genesis_block_returns_true_if_genesis(b):
 
 def test_lazy_execution():
     from bigchaindb.utils import Lazy
+
     l = Lazy()
     l.split(',')[1].split(' ').pop(1).strip()
     result = l.run('Like humans, cats tend to favor one paw over another')
     assert result == 'cats'
+
+    class Cat:
+        def __init__(self, name):
+            self.name = name
+
+    cat = Cat('Shmui')
+
+    l = Lazy()
+    l.name.upper()
+    result = l.run(cat)
+    assert result == 'SHMUI'

From 2c87f1c28acd81ebc25e5688d70307ac6b034938 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Fri, 27 Jan 2017 14:35:37 +0100
Subject: [PATCH 05/19] Update docs

---
 bigchaindb/backend/mongodb/connection.py |  5 +++++
 bigchaindb/utils.py                      | 15 +++++++++++++++
 2 files changed, 20 insertions(+)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 61af4469..2be39194 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -65,6 +65,11 @@ class MongoDBConnection(Connection):
 
 
 def collection(name):
+    """Return a lazy object that can be used to compose a query.
+
+    Args:
+        name (str): the name of the collection to query.
+    """
     return Lazy()[name]
 
 
diff --git a/bigchaindb/utils.py b/bigchaindb/utils.py
index 6ceace6b..1860dd3e 100644
--- a/bigchaindb/utils.py
+++ b/bigchaindb/utils.py
@@ -160,8 +160,17 @@ def is_genesis_block(block):
 
 
 class Lazy:
+    """Lazy objects are useful to create chains of methods to
+    execute later.
+
+    A lazy object records the methods that has been called, and
+    replay them when the :py:meth:`run` method is called. Note that
+    :py:meth:`run` needs an object `instance` to replay all the
+    methods that have been recorded.
+    """
 
     def __init__(self):
+        """Instantiate a new Lazy object."""
         self.stack = []
 
     def __getattr__(self, name):
@@ -178,6 +187,12 @@ class Lazy:
         return self
 
     def run(self, instance):
+        """Run the recorded chain of methods on `instance`.
+
+        Args:
+            instance: an object.
+        """
+
         last = instance
 
         for item in self.stack:

From a8bbc87c1ce0a4ceb6127368a71d23c7eef1e4ac Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Tue, 31 Jan 2017 01:27:55 +0100
Subject: [PATCH 06/19] Major improvs for MongoDBConnection class

---
 bigchaindb/backend/mongodb/connection.py | 73 +++++++++++++++++-------
 tests/backend/mongodb/test_connection.py |  4 +-
 2 files changed, 55 insertions(+), 22 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 2be39194..c641d726 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -1,5 +1,6 @@
 import time
 import logging
+from itertools import repeat
 
 from pymongo import MongoClient
 from pymongo import errors
@@ -12,9 +13,17 @@ from bigchaindb.backend.connection import Connection
 logger = logging.getLogger(__name__)
 
 
+# TODO: waiting for #1082 to be merged
+#       to move this constants in the configuration.
+
+CONNECTION_TIMEOUT = 4000  # in milliseconds
+MAX_RETRIES = 3  # number of tries before giving up, if 0 then try forever
+
+
 class MongoDBConnection(Connection):
 
-    def __init__(self, host=None, port=None, dbname=None, max_tries=3,
+    def __init__(self, host=None, port=None, dbname=None,
+                 connection_timeout=None, max_tries=None,
                  replicaset=None):
         """Create a new Connection instance.
 
@@ -22,7 +31,10 @@ class MongoDBConnection(Connection):
             host (str, optional): the host to connect to.
             port (int, optional): the port to connect to.
             dbname (str, optional): the database to use.
-            max_tries (int, optional): how many tries before giving up.
+            connection_timeout (int, optional): the milliseconds to wait
+                until timing out the database connection attempt.
+            max_tries (int, optional): how many tries before giving up,
+                if 0 then try forever.
             replicaset (str, optional): the name of the replica set to
                                         connect to.
         """
@@ -31,13 +43,15 @@ class MongoDBConnection(Connection):
         self.port = port or bigchaindb.config['database']['port']
         self.replicaset = replicaset or bigchaindb.config['database']['replicaset']
         self.dbname = dbname or bigchaindb.config['database']['name']
-        self.max_tries = max_tries
+        self.connection_timeout = connection_timeout if connection_timeout is not None else CONNECTION_TIMEOUT
+        self.max_tries = max_tries if max_tries is not None else MAX_RETRIES
+        self.max_tries_counter = range(self.max_tries) if self.max_tries != 0 else repeat(0)
         self.connection = None
 
     @property
     def conn(self):
         if self.connection is None:
-            self._connect()
+            self.connection = self._connect()
         return self.connection
 
     @property
@@ -45,23 +59,41 @@ class MongoDBConnection(Connection):
         return self.conn[self.dbname]
 
     def run(self, query):
-        return query.run(self.db)
+        attempt = 0
+        for i in self.max_tries_counter:
+            attempt += 1
+            try:
+                return query.run(self.conn[self.dbname])
+            except errors.AutoReconnect:
+                if attempt == self.max_tries:
+                    raise
+                self._connect()
 
     def _connect(self):
-        # we should only return a connection if the replica set is
-        # initialized. initialize_replica_set will check if the
-        # replica set is initialized else it will initialize it.
-        initialize_replica_set()
+        attempt = 0
+        for i in self.max_tries_counter:
+            attempt += 1
 
-        for i in range(self.max_tries):
             try:
-                self.connection = MongoClient(self.host, self.port,
-                                              replicaset=self.replicaset)
-            except errors.ConnectionFailure:
-                if i + 1 == self.max_tries:
-                    raise
-                else:
-                    time.sleep(2**i)
+                # FYI: this might raise a `ServerSelectionTimeoutError`,
+                # that is a subclass of `ConnectionFailure`.
+                connection = MongoClient(self.host,
+                                         self.port,
+                                         replicaset=self.replicaset,
+                                         serverselectiontimeoutms=self.connection_timeout)
+
+                # we should only return a connection if the replica set is
+                # initialized. initialize_replica_set will check if the
+                # replica set is initialized else it will initialize it.
+                initialize_replica_set(self.host, self.port, self.connection_timeout)
+                return connection
+            except (errors.ConnectionFailure, errors.AutoReconnect) as exc:
+                logger.warning('Attempt %s/%s. Connection to %s:%s failed after %sms.',
+                               attempt, self.max_tries if self.max_tries != 0 else '∞',
+                               self.host, self.port, self.connection_timeout)
+                if attempt == self.max_tries:
+                    logger.exception('Cannot connect to the Database. Giving up.')
+                    raise errors.ConnectionFailure() from exc
 
 
 def collection(name):
@@ -73,15 +105,16 @@ def collection(name):
     return Lazy()[name]
 
 
-def initialize_replica_set():
+def initialize_replica_set(host, port, connection_timeout):
     """Initialize a replica set. If already initialized skip."""
 
     # Setup a MongoDB connection
     # The reason we do this instead of `backend.connect` is that
     # `backend.connect` will connect you to a replica set but this fails if
     # you try to connect to a replica set that is not yet initialized
-    conn = MongoClient(host=bigchaindb.config['database']['host'],
-                       port=bigchaindb.config['database']['port'])
+    conn = MongoClient(host=host,
+                       port=port,
+                       serverselectiontimeoutms=connection_timeout)
     _check_replica_set(conn)
     host = '{}:{}'.format(bigchaindb.config['database']['host'],
                           bigchaindb.config['database']['port'])
diff --git a/tests/backend/mongodb/test_connection.py b/tests/backend/mongodb/test_connection.py
index 4c2919fe..49ab0604 100644
--- a/tests/backend/mongodb/test_connection.py
+++ b/tests/backend/mongodb/test_connection.py
@@ -138,7 +138,7 @@ def test_initialize_replica_set(mock_cmd_line_opts):
         ]
 
         # check that it returns
-        assert initialize_replica_set() is None
+        assert initialize_replica_set('host', 1337, 1000) is None
 
     # test it raises OperationError if anything wrong
     with mock.patch.object(Database, 'command') as mock_command:
@@ -148,4 +148,4 @@ def test_initialize_replica_set(mock_cmd_line_opts):
         ]
 
         with pytest.raises(OperationFailure):
-            initialize_replica_set()
+            initialize_replica_set('host', 1337, 1000)

From 1557645e946bb554a0877a267b0837308db9b590 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Tue, 31 Jan 2017 02:07:36 +0100
Subject: [PATCH 07/19] Better exception handling

---
 bigchaindb/backend/exceptions.py         |  3 +++
 bigchaindb/backend/mongodb/changefeed.py |  6 ++++--
 bigchaindb/backend/mongodb/connection.py | 18 ++++++++++--------
 bigchaindb/commands/bigchain.py          |  7 ++++++-
 4 files changed, 23 insertions(+), 11 deletions(-)

diff --git a/bigchaindb/backend/exceptions.py b/bigchaindb/backend/exceptions.py
index 1d55bc52..1f7c57a7 100644
--- a/bigchaindb/backend/exceptions.py
+++ b/bigchaindb/backend/exceptions.py
@@ -1,5 +1,8 @@
 from bigchaindb.exceptions import BigchainDBError
 
 
+class ConnectionError(BigchainDBError):
+    """Exception raised when the connection to the DataBase fails."""
+
 class DatabaseOpFailedError(BigchainDBError):
     """Exception for database operation errors."""
diff --git a/bigchaindb/backend/mongodb/changefeed.py b/bigchaindb/backend/mongodb/changefeed.py
index c54bd5da..ecf36bcc 100644
--- a/bigchaindb/backend/mongodb/changefeed.py
+++ b/bigchaindb/backend/mongodb/changefeed.py
@@ -8,7 +8,8 @@ from bigchaindb import backend
 from bigchaindb.backend.changefeed import ChangeFeed
 from bigchaindb.backend.utils import module_dispatch_registrar
 from bigchaindb.backend.mongodb.connection import MongoDBConnection
-
+from bigchaindb.backend.exceptions import (DatabaseOpFailedError,
+                                           ConnectionError)
 
 logger = logging.getLogger(__name__)
 register_changefeed = module_dispatch_registrar(backend.changefeed)
@@ -31,7 +32,8 @@ class MongoDBChangeFeed(ChangeFeed):
                 break
             except (errors.ConnectionFailure, errors.OperationFailure,
                     errors.AutoReconnect,
-                    errors.ServerSelectionTimeoutError) as exc:
+                    errors.ServerSelectionTimeoutError,
+                    DatabaseOpFailedError, ConnectionError) as exc:
                 logger.exception(exc)
                 time.sleep(1)
 
diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index c641d726..61cc430d 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -8,6 +8,7 @@ from pymongo import errors
 import bigchaindb
 from bigchaindb.utils import Lazy
 from bigchaindb.common import exceptions
+from bigchaindb.backend import exceptions as backend_exceptions
 from bigchaindb.backend.connection import Connection
 
 logger = logging.getLogger(__name__)
@@ -51,7 +52,7 @@ class MongoDBConnection(Connection):
     @property
     def conn(self):
         if self.connection is None:
-            self.connection = self._connect()
+            self._connect()
         return self.connection
 
     @property
@@ -77,23 +78,24 @@ class MongoDBConnection(Connection):
             try:
                 # FYI: this might raise a `ServerSelectionTimeoutError`,
                 # that is a subclass of `ConnectionFailure`.
-                connection = MongoClient(self.host,
-                                         self.port,
-                                         replicaset=self.replicaset,
-                                         serverselectiontimeoutms=self.connection_timeout)
+                self.connection = MongoClient(self.host,
+                                              self.port,
+                                              replicaset=self.replicaset,
+                                              serverselectiontimeoutms=self.connection_timeout)
 
                 # we should only return a connection if the replica set is
                 # initialized. initialize_replica_set will check if the
                 # replica set is initialized else it will initialize it.
                 initialize_replica_set(self.host, self.port, self.connection_timeout)
-                return connection
             except (errors.ConnectionFailure, errors.AutoReconnect) as exc:
                 logger.warning('Attempt %s/%s. Connection to %s:%s failed after %sms.',
                                attempt, self.max_tries if self.max_tries != 0 else '∞',
                                self.host, self.port, self.connection_timeout)
                 if attempt == self.max_tries:
-                    logger.exception('Cannot connect to the Database. Giving up.')
-                    raise errors.ConnectionFailure() from exc
+                    logger.critical('Cannot connect to the Database. Giving up.')
+                    raise backend_exceptions.ConnectionError() from exc
+            else:
+                break
 
 
 def collection(name):
diff --git a/bigchaindb/commands/bigchain.py b/bigchaindb/commands/bigchain.py
index 6661e902..4e18543d 100644
--- a/bigchaindb/commands/bigchain.py
+++ b/bigchaindb/commands/bigchain.py
@@ -23,7 +23,8 @@ from bigchaindb.utils import ProcessGroup
 from bigchaindb import backend
 from bigchaindb.backend import schema
 from bigchaindb.backend.admin import set_replicas, set_shards
-from bigchaindb.backend.exceptions import DatabaseOpFailedError
+from bigchaindb.backend.exceptions import (DatabaseOpFailedError,
+                                           ConnectionError)
 from bigchaindb.commands import utils
 from bigchaindb import processes
 
@@ -157,6 +158,8 @@ def run_init(args):
     except DatabaseAlreadyExists:
         print('The database already exists.', file=sys.stderr)
         print('If you wish to re-initialize it, first drop it.', file=sys.stderr)
+    except ConnectionError:
+        print('Cannot connect to the database.', file=sys.stderr)
 
 
 def run_drop(args):
@@ -201,6 +204,8 @@ def run_start(args):
         _run_init()
     except DatabaseAlreadyExists:
         pass
+    except ConnectionError:
+        print('Cannot connect to the database.', file=sys.stderr)
     except KeypairNotFoundException:
         sys.exit("Can't start BigchainDB, no keypair found. "
                  'Did you run `bigchaindb configure`?')

From 157db3e01fd08b0f0ba6770ed420e1b7931fb55f Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Tue, 31 Jan 2017 16:59:43 +0100
Subject: [PATCH 08/19] Fix exception in test

---
 tests/backend/mongodb/test_connection.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/tests/backend/mongodb/test_connection.py b/tests/backend/mongodb/test_connection.py
index 49ab0604..60181f25 100644
--- a/tests/backend/mongodb/test_connection.py
+++ b/tests/backend/mongodb/test_connection.py
@@ -56,13 +56,14 @@ def test_get_connection_returns_the_correct_instance():
 @mock.patch('time.sleep')
 def test_connection_error(mock_sleep, mock_client, mock_init_repl_set):
     from bigchaindb.backend import connect
+    from bigchaindb.backend.exceptions import ConnectionError
 
     # force the driver to trow ConnectionFailure
     # the mock on time.sleep is to prevent the actual sleep when running
     # the tests
     mock_client.side_effect = ConnectionFailure()
 
-    with pytest.raises(ConnectionFailure):
+    with pytest.raises(ConnectionError):
         conn = connect()
         conn.db
 

From 857cdb9b341568300fd8012149c70a747ca6c20f Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Tue, 31 Jan 2017 17:01:22 +0100
Subject: [PATCH 09/19] Checking for replica set is now within try..except

---
 bigchaindb/backend/mongodb/connection.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 61cc430d..79a085fb 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -76,17 +76,17 @@ class MongoDBConnection(Connection):
             attempt += 1
 
             try:
+                # we should only return a connection if the replica set is
+                # initialized. initialize_replica_set will check if the
+                # replica set is initialized else it will initialize it.
+                initialize_replica_set(self.host, self.port, self.connection_timeout)
+
                 # FYI: this might raise a `ServerSelectionTimeoutError`,
                 # that is a subclass of `ConnectionFailure`.
                 self.connection = MongoClient(self.host,
                                               self.port,
                                               replicaset=self.replicaset,
                                               serverselectiontimeoutms=self.connection_timeout)
-
-                # we should only return a connection if the replica set is
-                # initialized. initialize_replica_set will check if the
-                # replica set is initialized else it will initialize it.
-                initialize_replica_set(self.host, self.port, self.connection_timeout)
             except (errors.ConnectionFailure, errors.AutoReconnect) as exc:
                 logger.warning('Attempt %s/%s. Connection to %s:%s failed after %sms.',
                                attempt, self.max_tries if self.max_tries != 0 else '∞',

From 1588681c5b7edabf9e6b02206053a557e3233c73 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Tue, 31 Jan 2017 17:13:26 +0100
Subject: [PATCH 10/19] Fix pep8 error

---
 bigchaindb/backend/exceptions.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/bigchaindb/backend/exceptions.py b/bigchaindb/backend/exceptions.py
index 1f7c57a7..41bac0c6 100644
--- a/bigchaindb/backend/exceptions.py
+++ b/bigchaindb/backend/exceptions.py
@@ -4,5 +4,6 @@ from bigchaindb.exceptions import BigchainDBError
 class ConnectionError(BigchainDBError):
     """Exception raised when the connection to the DataBase fails."""
 
+
 class DatabaseOpFailedError(BigchainDBError):
     """Exception for database operation errors."""

From 16571b539f7e9034fbcbc8ca63d53b6b799f6022 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 2 Feb 2017 19:18:47 +0100
Subject: [PATCH 11/19] Normalize exceptions

---
 bigchaindb/backend/connection.py         |  8 +++
 bigchaindb/backend/exceptions.py         | 16 +++--
 bigchaindb/backend/mongodb/changefeed.py | 21 ++++---
 bigchaindb/backend/mongodb/connection.py | 74 ++++++++++++++----------
 bigchaindb/backend/mongodb/query.py      | 45 +++++++-------
 bigchaindb/backend/rethinkdb/admin.py    |  6 +-
 bigchaindb/commands/bigchain.py          |  6 +-
 tests/backend/mongodb/test_changefeed.py |  6 +-
 tests/backend/mongodb/test_connection.py |  2 +-
 tests/backend/rethinkdb/test_admin.py    |  8 +--
 10 files changed, 109 insertions(+), 83 deletions(-)

diff --git a/bigchaindb/backend/connection.py b/bigchaindb/backend/connection.py
index 0fda4078..bb4688da 100644
--- a/bigchaindb/backend/connection.py
+++ b/bigchaindb/backend/connection.py
@@ -32,6 +32,7 @@ def connect(backend=None, host=None, port=None, name=None, replicaset=None):
         based on the given (or defaulted) :attr:`backend`.
 
     Raises:
+        :exc:`~ConnectionError`: If the connection to the database fails.
         :exc:`~ConfigurationError`: If the given (or defaulted) :attr:`backend`
             is not supported or could not be loaded.
     """
@@ -77,6 +78,13 @@ class Connection:
 
         Args:
             query: the query to run
+        Raises:
+            :exc:`~DuplicateKeyError`: If the query fails because of a
+                duplicate key constraint.
+            :exc:`~OperationFailure`: If the query fails for any other
+                reason.
+            :exc:`~ConnectionError`: If the connection to the database
+                fails.
         """
 
         raise NotImplementedError()
diff --git a/bigchaindb/backend/exceptions.py b/bigchaindb/backend/exceptions.py
index 41bac0c6..017e19e4 100644
--- a/bigchaindb/backend/exceptions.py
+++ b/bigchaindb/backend/exceptions.py
@@ -1,9 +1,17 @@
 from bigchaindb.exceptions import BigchainDBError
 
 
-class ConnectionError(BigchainDBError):
-    """Exception raised when the connection to the DataBase fails."""
+class BackendError(BigchainDBError):
+    """Top level exception for any backend exception."""
 
 
-class DatabaseOpFailedError(BigchainDBError):
-    """Exception for database operation errors."""
+class ConnectionError(BackendError):
+    """Exception raised when the connection to the backend fails."""
+
+
+class OperationError(BackendError):
+    """Exception raised when a backend operation fails."""
+
+
+class DuplicateKeyError(OperationError):
+    """Exception raised when an insert fails because the key is not unique"""
diff --git a/bigchaindb/backend/mongodb/changefeed.py b/bigchaindb/backend/mongodb/changefeed.py
index ecf36bcc..0cf37943 100644
--- a/bigchaindb/backend/mongodb/changefeed.py
+++ b/bigchaindb/backend/mongodb/changefeed.py
@@ -1,15 +1,15 @@
+import os
 import logging
 import time
 
 import pymongo
-from pymongo import errors
 
 from bigchaindb import backend
 from bigchaindb.backend.changefeed import ChangeFeed
 from bigchaindb.backend.utils import module_dispatch_registrar
 from bigchaindb.backend.mongodb.connection import MongoDBConnection
-from bigchaindb.backend.exceptions import (DatabaseOpFailedError,
-                                           ConnectionError)
+from bigchaindb.backend.exceptions import BackendError
+
 
 logger = logging.getLogger(__name__)
 register_changefeed = module_dispatch_registrar(backend.changefeed)
@@ -30,11 +30,8 @@ class MongoDBChangeFeed(ChangeFeed):
             try:
                 self.run_changefeed()
                 break
-            except (errors.ConnectionFailure, errors.OperationFailure,
-                    errors.AutoReconnect,
-                    errors.ServerSelectionTimeoutError,
-                    DatabaseOpFailedError, ConnectionError) as exc:
-                logger.exception(exc)
+            except BackendError:
+                logger.exception('Error connecting to the database, retrying')
                 time.sleep(1)
 
     def run_changefeed(self):
@@ -43,9 +40,10 @@ class MongoDBChangeFeed(ChangeFeed):
         namespace = '{}.{}'.format(dbname, table)
         # last timestamp in the oplog. We only care for operations happening
         # in the future.
-        last_ts = self.connection.conn.local.oplog.rs.find()\
-                      .sort('$natural', pymongo.DESCENDING).limit(1)\
-                      .next()['ts']
+        last_ts = self.connection.run(
+            self.connection.query().local.oplog.rs.find()
+            .sort('$natural', pymongo.DESCENDING).limit(1)
+            .next()['ts'])
         # tailable cursor. A tailable cursor will remain open even after the
         # last result was returned. ``TAILABLE_AWAIT`` will block for some
         # timeout after the last result was returned. If no result is received
@@ -56,6 +54,7 @@ class MongoDBChangeFeed(ChangeFeed):
         )
 
         while cursor.alive:
+            print(os.getpid(), 'alive')
             try:
                 record = cursor.next()
             except StopIteration:
diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 79a085fb..5bd0d017 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -2,13 +2,14 @@ import time
 import logging
 from itertools import repeat
 
-from pymongo import MongoClient
-from pymongo import errors
+import pymongo
 
 import bigchaindb
 from bigchaindb.utils import Lazy
-from bigchaindb.common import exceptions
-from bigchaindb.backend import exceptions as backend_exceptions
+from bigchaindb.common.exceptions import ConfigurationError
+from bigchaindb.backend.exceptions import (DuplicateKeyError,
+                                           OperationError,
+                                           ConnectionError)
 from bigchaindb.backend.connection import Connection
 
 logger = logging.getLogger(__name__)
@@ -59,18 +60,33 @@ class MongoDBConnection(Connection):
     def db(self):
         return self.conn[self.dbname]
 
+    def query(self):
+        return Lazy()
+
+    def collection(self, name):
+        """Return a lazy object that can be used to compose a query.
+
+        Args:
+            name (str): the name of the collection to query.
+        """
+        return self.query()[self.dbname][name]
+
     def run(self, query):
-        attempt = 0
-        for i in self.max_tries_counter:
-            attempt += 1
-            try:
-                return query.run(self.conn[self.dbname])
-            except errors.AutoReconnect:
-                if attempt == self.max_tries:
-                    raise
-                self._connect()
+        try:
+            return query.run(self.conn)
+        except pymongo.errors.DuplicateKeyError as exc:
+            raise DuplicateKeyError from exc
+        except pymongo.errors.OperationFailure as exc:
+            raise OperationError from exc
 
     def _connect(self):
+        """Try to connect to the database.
+
+        Raises:
+            :exc:`~ConnectionError`: If the connection to the database
+                fails.
+        """
+
         attempt = 0
         for i in self.max_tries_counter:
             attempt += 1
@@ -83,30 +99,24 @@ class MongoDBConnection(Connection):
 
                 # FYI: this might raise a `ServerSelectionTimeoutError`,
                 # that is a subclass of `ConnectionFailure`.
-                self.connection = MongoClient(self.host,
-                                              self.port,
-                                              replicaset=self.replicaset,
-                                              serverselectiontimeoutms=self.connection_timeout)
-            except (errors.ConnectionFailure, errors.AutoReconnect) as exc:
+                self.connection = pymongo.MongoClient(self.host,
+                                                      self.port,
+                                                      replicaset=self.replicaset,
+                                                      serverselectiontimeoutms=self.connection_timeout)
+
+            # `initialize_replica_set` might raise `ConnectionFailure` or `OperationFailure`.
+            except (pymongo.errors.ConnectionFailure,
+                    pymongo.errors.OperationFailure) as exc:
                 logger.warning('Attempt %s/%s. Connection to %s:%s failed after %sms.',
                                attempt, self.max_tries if self.max_tries != 0 else '∞',
                                self.host, self.port, self.connection_timeout)
                 if attempt == self.max_tries:
                     logger.critical('Cannot connect to the Database. Giving up.')
-                    raise backend_exceptions.ConnectionError() from exc
+                    raise ConnectionError() from exc
             else:
                 break
 
 
-def collection(name):
-    """Return a lazy object that can be used to compose a query.
-
-    Args:
-        name (str): the name of the collection to query.
-    """
-    return Lazy()[name]
-
-
 def initialize_replica_set(host, port, connection_timeout):
     """Initialize a replica set. If already initialized skip."""
 
@@ -114,7 +124,7 @@ def initialize_replica_set(host, port, connection_timeout):
     # The reason we do this instead of `backend.connect` is that
     # `backend.connect` will connect you to a replica set but this fails if
     # you try to connect to a replica set that is not yet initialized
-    conn = MongoClient(host=host,
+    conn = pymongo.MongoClient(host=host,
                        port=port,
                        serverselectiontimeoutms=connection_timeout)
     _check_replica_set(conn)
@@ -125,7 +135,7 @@ def initialize_replica_set(host, port, connection_timeout):
 
     try:
         conn.admin.command('replSetInitiate', config)
-    except errors.OperationFailure as exc_info:
+    except pymongo.errors.OperationFailure as exc_info:
         if exc_info.details['codeName'] == 'AlreadyInitialized':
             return
         raise
@@ -153,12 +163,12 @@ def _check_replica_set(conn):
         repl_opts = options['parsed']['replication']
         repl_set_name = repl_opts.get('replSetName', None) or repl_opts['replSet']
     except KeyError:
-        raise exceptions.ConfigurationError('mongod was not started with'
+        raise ConfigurationError('mongod was not started with'
                                             ' the replSet option.')
 
     bdb_repl_set_name = bigchaindb.config['database']['replicaset']
     if repl_set_name != bdb_repl_set_name:
-        raise exceptions.ConfigurationError('The replicaset configuration of '
+        raise ConfigurationError('The replicaset configuration of '
                                             'bigchaindb (`{}`) needs to match '
                                             'the replica set name from MongoDB'
                                             ' (`{}`)'
diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py
index b4739dc2..e68608d7 100644
--- a/bigchaindb/backend/mongodb/query.py
+++ b/bigchaindb/backend/mongodb/query.py
@@ -10,8 +10,9 @@ from pymongo import errors
 from bigchaindb import backend
 from bigchaindb.common.exceptions import CyclicBlockchainError
 from bigchaindb.common.transaction import Transaction
+from bigchaindb.backend.exceptions import DuplicateKeyError
 from bigchaindb.backend.utils import module_dispatch_registrar
-from bigchaindb.backend.mongodb.connection import MongoDBConnection, collection
+from bigchaindb.backend.mongodb.connection import MongoDBConnection
 
 
 register_query = module_dispatch_registrar(backend.query)
@@ -21,9 +22,9 @@ register_query = module_dispatch_registrar(backend.query)
 def write_transaction(conn, signed_transaction):
     try:
         return conn.run(
-            collection('backlog')
+            conn.collection('backlog')
             .insert_one(signed_transaction))
-    except errors.DuplicateKeyError:
+    except DuplicateKeyError:
         return
 
 
@@ -32,7 +33,7 @@ def update_transaction(conn, transaction_id, doc):
     # with mongodb we need to add update operators to the doc
     doc = {'$set': doc}
     return conn.run(
-        collection('backlog')
+        conn.collection('backlog')
         .find_one_and_update(
             {'id': transaction_id},
             doc,
@@ -42,14 +43,14 @@ def update_transaction(conn, transaction_id, doc):
 @register_query(MongoDBConnection)
 def delete_transaction(conn, *transaction_id):
     return conn.run(
-        collection('backlog')
+        conn.collection('backlog')
         .delete_many({'id': {'$in': transaction_id}}))
 
 
 @register_query(MongoDBConnection)
 def get_stale_transactions(conn, reassign_delay):
     return conn.run(
-        collection('backlog')
+        conn.collection('backlog')
         .find({'assignment_timestamp': {'$lt': time() - reassign_delay}},
               projection={'_id': False}))
 
@@ -58,7 +59,7 @@ def get_stale_transactions(conn, reassign_delay):
 def get_transaction_from_block(conn, transaction_id, block_id):
     try:
         return conn.run(
-            collection('bigchain')
+            conn.collection('bigchain')
             .aggregate([
                 {'$match': {'id': block_id}},
                 {'$project': {
@@ -84,7 +85,7 @@ def get_transaction_from_block(conn, transaction_id, block_id):
 @register_query(MongoDBConnection)
 def get_transaction_from_backlog(conn, transaction_id):
     return conn.run(
-        collection('backlog')
+        conn.collection('backlog')
         .find_one({'id': transaction_id},
                   projection={'_id': False,
                               'assignee': False,
@@ -94,7 +95,7 @@ def get_transaction_from_backlog(conn, transaction_id):
 @register_query(MongoDBConnection)
 def get_blocks_status_from_transaction(conn, transaction_id):
     return conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .find({'block.transactions.id': transaction_id},
               projection=['id', 'block.voters']))
 
@@ -139,7 +140,7 @@ def get_txids_filtered(conn, asset_id, operation=None):
 @register_query(MongoDBConnection)
 def get_asset_by_id(conn, asset_id):
     cursor = conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .aggregate([
             {'$match': {
                 'block.transactions.id': asset_id,
@@ -160,7 +161,7 @@ def get_asset_by_id(conn, asset_id):
 @register_query(MongoDBConnection)
 def get_spent(conn, transaction_id, output):
     cursor = conn.run(
-        collection('bigchain').aggregate([
+        conn.collection('bigchain').aggregate([
             {'$unwind': '$block.transactions'},
             {'$match': {
                 'block.transactions.inputs.fulfills.txid': transaction_id,
@@ -175,7 +176,7 @@ def get_spent(conn, transaction_id, output):
 @register_query(MongoDBConnection)
 def get_owned_ids(conn, owner):
     cursor = conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .aggregate([
             {'$unwind': '$block.transactions'},
             {'$match': {
@@ -192,7 +193,7 @@ def get_owned_ids(conn, owner):
 @register_query(MongoDBConnection)
 def get_votes_by_block_id(conn, block_id):
     return conn.run(
-        collection('votes')
+        conn.collection('votes')
         .find({'vote.voting_for_block': block_id},
               projection={'_id': False}))
 
@@ -200,7 +201,7 @@ def get_votes_by_block_id(conn, block_id):
 @register_query(MongoDBConnection)
 def get_votes_by_block_id_and_voter(conn, block_id, node_pubkey):
     return conn.run(
-        collection('votes')
+        conn.collection('votes')
         .find({'vote.voting_for_block': block_id,
                'node_pubkey': node_pubkey},
               projection={'_id': False}))
@@ -209,14 +210,14 @@ def get_votes_by_block_id_and_voter(conn, block_id, node_pubkey):
 @register_query(MongoDBConnection)
 def write_block(conn, block):
     return conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .insert_one(block.to_dict()))
 
 
 @register_query(MongoDBConnection)
 def get_block(conn, block_id):
     return conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .find_one({'id': block_id},
                   projection={'_id': False}))
 
@@ -224,27 +225,27 @@ def get_block(conn, block_id):
 @register_query(MongoDBConnection)
 def has_transaction(conn, transaction_id):
     return bool(conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .find_one({'block.transactions.id': transaction_id})))
 
 
 @register_query(MongoDBConnection)
 def count_blocks(conn):
     return conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .count())
 
 
 @register_query(MongoDBConnection)
 def count_backlog(conn):
     return conn.run(
-        collection('backlog')
+        conn.collection('backlog')
         .count())
 
 
 @register_query(MongoDBConnection)
 def write_vote(conn, vote):
-    conn.run(collection('votes').insert_one(vote))
+    conn.run(conn.collection('votes').insert_one(vote))
     vote.pop('_id')
     return vote
 
@@ -252,7 +253,7 @@ def write_vote(conn, vote):
 @register_query(MongoDBConnection)
 def get_genesis_block(conn):
     return conn.run(
-        collection('bigchain')
+        conn.collection('bigchain')
         .find_one(
             {'block.transactions.0.operation': 'GENESIS'},
             {'_id': False}
@@ -262,7 +263,7 @@ def get_genesis_block(conn):
 @register_query(MongoDBConnection)
 def get_last_voted_block(conn, node_pubkey):
     last_voted = conn.run(
-            collection('votes')
+            conn.collection('votes')
             .find({'node_pubkey': node_pubkey},
                   sort=[('vote.timestamp', -1)]))
 
diff --git a/bigchaindb/backend/rethinkdb/admin.py b/bigchaindb/backend/rethinkdb/admin.py
index 63548f87..23b55048 100644
--- a/bigchaindb/backend/rethinkdb/admin.py
+++ b/bigchaindb/backend/rethinkdb/admin.py
@@ -5,7 +5,7 @@ import rethinkdb as r
 
 from bigchaindb.backend import admin
 from bigchaindb.backend.schema import TABLES
-from bigchaindb.backend.exceptions import DatabaseOpFailedError
+from bigchaindb.backend.exceptions import OperationError
 from bigchaindb.backend.utils import module_dispatch_registrar
 from bigchaindb.backend.rethinkdb.connection import RethinkDBConnection
 
@@ -78,7 +78,7 @@ def reconfigure(connection, *, table, shards, replicas,
             <https://rethinkdb.com/api/python/reconfigure/>`_.
 
     Raises:
-        DatabaseOpFailedError: If the reconfiguration fails due to a
+        OperationError: If the reconfiguration fails due to a
             RethinkDB :exc:`ReqlOpFailedError` or
             :exc:`ReqlQueryLogicError`.
 
@@ -96,7 +96,7 @@ def reconfigure(connection, *, table, shards, replicas,
     try:
         return connection.run(r.table(table).reconfigure(**params))
     except (r.ReqlOpFailedError, r.ReqlQueryLogicError) as e:
-        raise DatabaseOpFailedError from e
+        raise OperationError from e
 
 
 @register_admin(RethinkDBConnection)
diff --git a/bigchaindb/commands/bigchain.py b/bigchaindb/commands/bigchain.py
index 4e18543d..89b84dba 100644
--- a/bigchaindb/commands/bigchain.py
+++ b/bigchaindb/commands/bigchain.py
@@ -23,7 +23,7 @@ from bigchaindb.utils import ProcessGroup
 from bigchaindb import backend
 from bigchaindb.backend import schema
 from bigchaindb.backend.admin import set_replicas, set_shards
-from bigchaindb.backend.exceptions import (DatabaseOpFailedError,
+from bigchaindb.backend.exceptions import (OperationError,
                                            ConnectionError)
 from bigchaindb.commands import utils
 from bigchaindb import processes
@@ -252,7 +252,7 @@ def run_set_shards(args):
     conn = backend.connect()
     try:
         set_shards(conn, shards=args.num_shards)
-    except DatabaseOpFailedError as e:
+    except OperationError as e:
         logger.warn(e)
 
 
@@ -260,7 +260,7 @@ def run_set_replicas(args):
     conn = backend.connect()
     try:
         set_replicas(conn, replicas=args.num_replicas)
-    except DatabaseOpFailedError as e:
+    except OperationError as e:
         logger.warn(e)
 
 
diff --git a/tests/backend/mongodb/test_changefeed.py b/tests/backend/mongodb/test_changefeed.py
index e7581b34..de07613c 100644
--- a/tests/backend/mongodb/test_changefeed.py
+++ b/tests/backend/mongodb/test_changefeed.py
@@ -1,7 +1,6 @@
 from unittest import mock
 
 import pytest
-from pymongo.errors import ConnectionFailure
 
 from multipipes import Pipe
 
@@ -152,14 +151,15 @@ def test_changefeed_prefeed(mock_cursor_next, mock_cursor_alive,
 
 @pytest.mark.bdb
 @mock.patch('pymongo.cursor.Cursor.alive', new_callable=mock.PropertyMock)
-@mock.patch('bigchaindb.backend.mongodb.changefeed.MongoDBChangeFeed.run_changefeed')  # noqa
+@mock.patch('bigchaindb.backend.mongodb.connection.MongoDBConnection.run')  # noqa
 def test_connection_failure(mock_run_changefeed, mock_cursor_alive):
     from bigchaindb.backend import get_changefeed, connect
+    from bigchaindb.backend.exceptions import ConnectionError
     from bigchaindb.backend.changefeed import ChangeFeed
 
     conn = connect()
     mock_cursor_alive.return_value = False
-    mock_run_changefeed.side_effect = [ConnectionFailure(), mock.DEFAULT]
+    mock_run_changefeed.side_effect = [ConnectionError(), mock.DEFAULT]
 
     changefeed = get_changefeed(conn, 'backlog', ChangeFeed.INSERT)
     changefeed.run_forever()
diff --git a/tests/backend/mongodb/test_connection.py b/tests/backend/mongodb/test_connection.py
index 60181f25..feda77ed 100644
--- a/tests/backend/mongodb/test_connection.py
+++ b/tests/backend/mongodb/test_connection.py
@@ -58,7 +58,7 @@ def test_connection_error(mock_sleep, mock_client, mock_init_repl_set):
     from bigchaindb.backend import connect
     from bigchaindb.backend.exceptions import ConnectionError
 
-    # force the driver to trow ConnectionFailure
+    # force the driver to throw ConnectionFailure
     # the mock on time.sleep is to prevent the actual sleep when running
     # the tests
     mock_client.side_effect = ConnectionFailure()
diff --git a/tests/backend/rethinkdb/test_admin.py b/tests/backend/rethinkdb/test_admin.py
index 8c4f0528..0f71ca21 100644
--- a/tests/backend/rethinkdb/test_admin.py
+++ b/tests/backend/rethinkdb/test_admin.py
@@ -177,8 +177,8 @@ def test_reconfigure_replicas_without_nonvoting_replica_tags(rdb_conn,
                                                              db_name,
                                                              db_conn):
     from bigchaindb.backend.rethinkdb.admin import reconfigure
-    from bigchaindb.backend.exceptions import DatabaseOpFailedError
-    with pytest.raises(DatabaseOpFailedError) as exc:
+    from bigchaindb.backend.exceptions import OperationError
+    with pytest.raises(OperationError) as exc:
         reconfigure(db_conn, table='backlog', shards=1,
                     replicas={'default': 1}, primary_replica_tag='default')
     assert isinstance(exc.value.__cause__, r.ReqlQueryLogicError)
@@ -187,8 +187,8 @@ def test_reconfigure_replicas_without_nonvoting_replica_tags(rdb_conn,
 @pytest.mark.bdb
 def test_reconfigure_too_many_replicas(rdb_conn, db_name, db_conn):
     from bigchaindb.backend.rethinkdb.admin import reconfigure
-    from bigchaindb.backend.exceptions import DatabaseOpFailedError
+    from bigchaindb.backend.exceptions import OperationError
     replicas = _count_rethinkdb_servers() + 1
-    with pytest.raises(DatabaseOpFailedError) as exc:
+    with pytest.raises(OperationError) as exc:
         reconfigure(db_conn, table='backlog', shards=1, replicas=replicas)
     assert isinstance(exc.value.__cause__, r.ReqlOpFailedError)

From e3a6d3d3433e99441b4fd3f5eadf27fa2e7ae2ab Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 2 Feb 2017 19:58:00 +0100
Subject: [PATCH 12/19] Add tests for connection.run

---
 bigchaindb/backend/mongodb/connection.py |  2 ++
 tests/backend/mongodb/test_connection.py | 32 +++++++++++++++++++++---
 2 files changed, 30 insertions(+), 4 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index ce0c6113..4bd822b4 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -76,6 +76,8 @@ class MongoDBConnection(Connection):
             return query.run(self.conn)
         except pymongo.errors.DuplicateKeyError as exc:
             raise DuplicateKeyError from exc
+        except pymongo.errors.AutoReconnect as exc:
+            raise ConnectionError from exc
         except pymongo.errors.OperationFailure as exc:
             raise OperationError from exc
 
diff --git a/tests/backend/mongodb/test_connection.py b/tests/backend/mongodb/test_connection.py
index feda77ed..be660d0a 100644
--- a/tests/backend/mongodb/test_connection.py
+++ b/tests/backend/mongodb/test_connection.py
@@ -1,9 +1,9 @@
 from unittest import mock
 
 import pytest
+import pymongo
 from pymongo import MongoClient
 from pymongo.database import Database
-from pymongo.errors import ConnectionFailure, OperationFailure
 
 
 pytestmark = pytest.mark.bdb
@@ -61,7 +61,7 @@ def test_connection_error(mock_sleep, mock_client, mock_init_repl_set):
     # force the driver to throw ConnectionFailure
     # the mock on time.sleep is to prevent the actual sleep when running
     # the tests
-    mock_client.side_effect = ConnectionFailure()
+    mock_client.side_effect = pymongo.errors.ConnectionFailure()
 
     with pytest.raises(ConnectionError):
         conn = connect()
@@ -70,6 +70,30 @@ def test_connection_error(mock_sleep, mock_client, mock_init_repl_set):
     assert mock_client.call_count == 3
 
 
+@mock.patch('bigchaindb.backend.mongodb.connection.initialize_replica_set')
+@mock.patch('pymongo.MongoClient')
+def test_connection_run_errors(mock_client, mock_init_repl_set):
+    from bigchaindb.backend import connect
+    from bigchaindb.backend.exceptions import (DuplicateKeyError,
+                                               OperationError,
+                                               ConnectionError)
+
+    conn = connect()
+    query = mock.Mock()
+
+    query.run.side_effect = pymongo.errors.AutoReconnect('foo')
+    with pytest.raises(ConnectionError):
+        conn.run(query)
+
+    query.run.side_effect = pymongo.errors.DuplicateKeyError('foo')
+    with pytest.raises(DuplicateKeyError):
+        conn.run(query)
+
+    query.run.side_effect = pymongo.errors.OperationFailure('foo')
+    with pytest.raises(OperationError):
+        conn.run(query)
+
+
 def test_check_replica_set_not_enabled(mongodb_connection):
     from bigchaindb.backend.mongodb.connection import _check_replica_set
     from bigchaindb.common.exceptions import ConfigurationError
@@ -145,8 +169,8 @@ def test_initialize_replica_set(mock_cmd_line_opts):
     with mock.patch.object(Database, 'command') as mock_command:
         mock_command.side_effect = [
             mock_cmd_line_opts,
-            OperationFailure(None, details={'codeName': ''})
+            pymongo.errors.OperationFailure(None, details={'codeName': ''})
         ]
 
-        with pytest.raises(OperationFailure):
+        with pytest.raises(pymongo.errors.OperationFailure):
             initialize_replica_set('host', 1337, 1000)

From 6d3c04169c3160daedef9421462052f4e2a9792a Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 2 Feb 2017 20:11:31 +0100
Subject: [PATCH 13/19] Remove except for connection error in commands

It's out of scoper for this PR
---
 bigchaindb/commands/bigchain.py | 7 +------
 1 file changed, 1 insertion(+), 6 deletions(-)

diff --git a/bigchaindb/commands/bigchain.py b/bigchaindb/commands/bigchain.py
index 1743b851..c8a5ba72 100644
--- a/bigchaindb/commands/bigchain.py
+++ b/bigchaindb/commands/bigchain.py
@@ -24,8 +24,7 @@ from bigchaindb import backend
 from bigchaindb.backend import schema
 from bigchaindb.backend.admin import (set_replicas, set_shards, add_replicas,
                                       remove_replicas)
-from bigchaindb.backend.exceptions import (OperationError,
-                                           ConnectionError)
+from bigchaindb.backend.exceptions import OperationError
 from bigchaindb.commands import utils
 from bigchaindb import processes
 
@@ -164,8 +163,6 @@ def run_init(args):
     except DatabaseAlreadyExists:
         print('The database already exists.', file=sys.stderr)
         print('If you wish to re-initialize it, first drop it.', file=sys.stderr)
-    except ConnectionError:
-        print('Cannot connect to the database.', file=sys.stderr)
 
 
 def run_drop(args):
@@ -210,8 +207,6 @@ def run_start(args):
         _run_init()
     except DatabaseAlreadyExists:
         pass
-    except ConnectionError:
-        print('Cannot connect to the database.', file=sys.stderr)
     except KeypairNotFoundException:
         sys.exit("Can't start BigchainDB, no keypair found. "
                  'Did you run `bigchaindb configure`?')

From 98135c7df9e43367b19da711c84977e5e15c3cc5 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Thu, 2 Feb 2017 20:59:17 +0100
Subject: [PATCH 14/19] Add hack to handle reconnection to changefeed

---
 bigchaindb/backend/mongodb/changefeed.py | 17 ++++++++++-------
 1 file changed, 10 insertions(+), 7 deletions(-)

diff --git a/bigchaindb/backend/mongodb/changefeed.py b/bigchaindb/backend/mongodb/changefeed.py
index 0cf37943..8bdaca92 100644
--- a/bigchaindb/backend/mongodb/changefeed.py
+++ b/bigchaindb/backend/mongodb/changefeed.py
@@ -1,4 +1,3 @@
-import os
 import logging
 import time
 
@@ -28,9 +27,13 @@ class MongoDBChangeFeed(ChangeFeed):
 
         while True:
             try:
+                # XXX: hack to force reconnection,
+                #      the correct way to fix this is to manage errors
+                #      for cursors.conn
+                self.connection.connection = None
                 self.run_changefeed()
                 break
-            except BackendError:
+            except (BackendError, pymongo.errors.ConnectionFailure):
                 logger.exception('Error connecting to the database, retrying')
                 time.sleep(1)
 
@@ -48,13 +51,13 @@ class MongoDBChangeFeed(ChangeFeed):
         # last result was returned. ``TAILABLE_AWAIT`` will block for some
         # timeout after the last result was returned. If no result is received
         # in the meantime it will raise a StopIteration excetiption.
-        cursor = self.connection.conn.local.oplog.rs.find(
-            {'ns': namespace, 'ts': {'$gt': last_ts}},
-            cursor_type=pymongo.CursorType.TAILABLE_AWAIT
-        )
+        cursor = self.connection.run(
+            self.connection.query().local.oplog.rs.find(
+                {'ns': namespace, 'ts': {'$gt': last_ts}},
+                cursor_type=pymongo.CursorType.TAILABLE_AWAIT
+            ))
 
         while cursor.alive:
-            print(os.getpid(), 'alive')
             try:
                 record = cursor.next()
             except StopIteration:

From 7f50d76d9ec5ac3d3f8a2f3d50cea6b6314e07f0 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Fri, 3 Feb 2017 11:58:18 +0100
Subject: [PATCH 15/19] Fix test on changefeed reconnection

---
 tests/backend/mongodb/test_changefeed.py | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/tests/backend/mongodb/test_changefeed.py b/tests/backend/mongodb/test_changefeed.py
index de07613c..67b54cd8 100644
--- a/tests/backend/mongodb/test_changefeed.py
+++ b/tests/backend/mongodb/test_changefeed.py
@@ -150,16 +150,15 @@ def test_changefeed_prefeed(mock_cursor_next, mock_cursor_alive,
 
 
 @pytest.mark.bdb
-@mock.patch('pymongo.cursor.Cursor.alive', new_callable=mock.PropertyMock)
-@mock.patch('bigchaindb.backend.mongodb.connection.MongoDBConnection.run')  # noqa
-def test_connection_failure(mock_run_changefeed, mock_cursor_alive):
+@mock.patch('bigchaindb.backend.mongodb.changefeed.MongoDBChangeFeed.run_changefeed')  # noqa
+def test_connection_failure(mock_run_changefeed):
     from bigchaindb.backend import get_changefeed, connect
     from bigchaindb.backend.exceptions import ConnectionError
     from bigchaindb.backend.changefeed import ChangeFeed
 
     conn = connect()
-    mock_cursor_alive.return_value = False
-    mock_run_changefeed.side_effect = [ConnectionError(), mock.DEFAULT]
+    mock_run_changefeed.side_effect = [ConnectionError(),
+                                       mock.DEFAULT]
 
     changefeed = get_changefeed(conn, 'backlog', ChangeFeed.INSERT)
     changefeed.run_forever()

From 3c45de70d020a796ec8cfb00da23993c594c704e Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Mon, 6 Feb 2017 14:43:40 +0100
Subject: [PATCH 16/19] Wrap queries with connection object

---
 bigchaindb/backend/mongodb/query.py | 36 ++++++++++++++++-------------
 1 file changed, 20 insertions(+), 16 deletions(-)

diff --git a/bigchaindb/backend/mongodb/query.py b/bigchaindb/backend/mongodb/query.py
index 5347df9f..6a241a78 100644
--- a/bigchaindb/backend/mongodb/query.py
+++ b/bigchaindb/backend/mongodb/query.py
@@ -121,7 +121,9 @@ def get_txids_filtered(conn, asset_id, operation=None):
         {'$match': match},
         {'$project': {'block.transactions.id': True}}
     ]
-    cursor = conn.db['bigchain'].aggregate(pipeline)
+    cursor = conn.run(
+        conn.collection('bigchain')
+        .aggregate(pipeline))
     return (elem['block']['transactions']['id'] for elem in cursor)
 
 
@@ -281,18 +283,20 @@ def get_last_voted_block(conn, node_pubkey):
 
 @register_query(MongoDBConnection)
 def get_unvoted_blocks(conn, node_pubkey):
-    return conn.db['bigchain'].aggregate([
-        {'$lookup': {
-            'from': 'votes',
-            'localField': 'id',
-            'foreignField': 'vote.voting_for_block',
-            'as': 'votes'
-        }},
-        {'$match': {
-            'votes.node_pubkey': {'$ne': node_pubkey},
-            'block.transactions.operation': {'$ne': 'GENESIS'}
-        }},
-        {'$project': {
-            'votes': False, '_id': False
-        }}
-    ])
+    return conn.run(
+        conn.collection('bigchain')
+        .aggregate([
+            {'$lookup': {
+                'from': 'votes',
+                'localField': 'id',
+                'foreignField': 'vote.voting_for_block',
+                'as': 'votes'
+            }},
+            {'$match': {
+                'votes.node_pubkey': {'$ne': node_pubkey},
+                'block.transactions.operation': {'$ne': 'GENESIS'}
+            }},
+            {'$project': {
+                'votes': False, '_id': False
+            }}
+        ]))

From 5604e32d572d3a375318679ed7ddf3882e4bf7cc Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Mon, 6 Feb 2017 16:14:14 +0100
Subject: [PATCH 17/19] Try once if AutoReconnect is raised

---
 bigchaindb/backend/mongodb/connection.py | 11 ++++++++---
 tests/backend/mongodb/test_connection.py |  7 ++++++-
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/bigchaindb/backend/mongodb/connection.py b/bigchaindb/backend/mongodb/connection.py
index 4bd822b4..d01d5861 100644
--- a/bigchaindb/backend/mongodb/connection.py
+++ b/bigchaindb/backend/mongodb/connection.py
@@ -73,11 +73,16 @@ class MongoDBConnection(Connection):
 
     def run(self, query):
         try:
-            return query.run(self.conn)
-        except pymongo.errors.DuplicateKeyError as exc:
-            raise DuplicateKeyError from exc
+            try:
+                return query.run(self.conn)
+            except pymongo.errors.AutoReconnect as exc:
+                logger.warning('Lost connection to the database, '
+                               'retrying query.')
+                return query.run(self.conn)
         except pymongo.errors.AutoReconnect as exc:
             raise ConnectionError from exc
+        except pymongo.errors.DuplicateKeyError as exc:
+            raise DuplicateKeyError from exc
         except pymongo.errors.OperationFailure as exc:
             raise OperationError from exc
 
diff --git a/tests/backend/mongodb/test_connection.py b/tests/backend/mongodb/test_connection.py
index be660d0a..786b7d7b 100644
--- a/tests/backend/mongodb/test_connection.py
+++ b/tests/backend/mongodb/test_connection.py
@@ -79,19 +79,24 @@ def test_connection_run_errors(mock_client, mock_init_repl_set):
                                                ConnectionError)
 
     conn = connect()
-    query = mock.Mock()
 
+    query = mock.Mock()
     query.run.side_effect = pymongo.errors.AutoReconnect('foo')
     with pytest.raises(ConnectionError):
         conn.run(query)
+    assert query.run.call_count == 2
 
+    query = mock.Mock()
     query.run.side_effect = pymongo.errors.DuplicateKeyError('foo')
     with pytest.raises(DuplicateKeyError):
         conn.run(query)
+    assert query.run.call_count == 1
 
+    query = mock.Mock()
     query.run.side_effect = pymongo.errors.OperationFailure('foo')
     with pytest.raises(OperationError):
         conn.run(query)
+    assert query.run.call_count == 1
 
 
 def test_check_replica_set_not_enabled(mongodb_connection):

From 0c8927dbbe80471f59241b556b24e31dc4396c3f Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Mon, 6 Feb 2017 16:21:22 +0100
Subject: [PATCH 18/19] Add comment to hackish solution

---
 bigchaindb/backend/mongodb/changefeed.py | 8 +++++---
 1 file changed, 5 insertions(+), 3 deletions(-)

diff --git a/bigchaindb/backend/mongodb/changefeed.py b/bigchaindb/backend/mongodb/changefeed.py
index 8bdaca92..4a5a5b7e 100644
--- a/bigchaindb/backend/mongodb/changefeed.py
+++ b/bigchaindb/backend/mongodb/changefeed.py
@@ -27,9 +27,11 @@ class MongoDBChangeFeed(ChangeFeed):
 
         while True:
             try:
-                # XXX: hack to force reconnection,
-                #      the correct way to fix this is to manage errors
-                #      for cursors.conn
+                # XXX: hack to force reconnection. Why? Because the cursor
+                # in `run_changefeed` does not run in the context of a
+                # Connection object, so if the connection is lost we need
+                # to manually reset the connection to None.
+                # See #1154
                 self.connection.connection = None
                 self.run_changefeed()
                 break

From 2962b4a27d45fb87fb8a7c2cc345bcb4727ffb12 Mon Sep 17 00:00:00 2001
From: vrde <agranzot@gmail.com>
Date: Mon, 6 Feb 2017 18:01:21 +0100
Subject: [PATCH 19/19] Skip test

---
 tests/backend/mongodb/test_indexes.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/tests/backend/mongodb/test_indexes.py b/tests/backend/mongodb/test_indexes.py
index ba6afae1..d11d124c 100644
--- a/tests/backend/mongodb/test_indexes.py
+++ b/tests/backend/mongodb/test_indexes.py
@@ -4,6 +4,7 @@ from unittest.mock import MagicMock
 pytestmark = pytest.mark.bdb
 
 
+@pytest.mark.skipif(reason='Will be handled by #1126')
 def test_asset_id_index():
     from bigchaindb.backend.mongodb.query import get_txids_filtered
     from bigchaindb.backend import connect