Flat UTXO collection and first integration with Tendermint™ (#1822)

* Remove testing for rethinkdb, mongodb, and Py3.5

* Add first tests

* Add validation

* Add command to start the ABCI Server

* Reuse existing MongoDB Connection class

* Use DuplicateTransaction

* Test only tendermint

* Update travis scripts

* Fix pep8 errors

* Update Makefile

.ci/travis-before-script.sh

@@ -2,30 +2,26 @@
 
 set -e -x
 
-if [[ "${BIGCHAINDB_DATABASE_BACKEND}" == rethinkdb ]]; then
-    docker pull rethinkdb:2.3.5
-    docker run -d --publish=28015:28015 --name rdb rethinkdb:2.3.5
-elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == mongodb && \
+if [[ "${BIGCHAINDB_DATABASE_BACKEND}" == localmongodb && \
         -z "${BIGCHAINDB_DATABASE_SSL}" ]]; then
     # Connect to MongoDB on port 27017 via a normal, unsecure connection if
     # BIGCHAINDB_DATABASE_SSL is unset.
     # It is unset in this case in .travis.yml.
-    docker pull mongo:3.4.4
-    docker run -d --publish=27017:27017 --name mdb-without-ssl mongo:3.4.4 \
-        --replSet=bigchain-rs
-elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == mongodb && \
+    docker pull mongo:3.4
+    docker run -d --publish=27017:27017 --name mdb-without-ssl mongo:3.4 # --replSet=bigchain-rs
+elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == localmongodb && \
         "${BIGCHAINDB_DATABASE_SSL}" == true ]]; then
     # Connect to MongoDB on port 27017 via TLS/SSL connection if
     # BIGCHAINDB_DATABASE_SSL is set.
     # It is set to 'true' here in .travis.yml. Dummy certificates for testing
     # are stored under bigchaindb/tests/backend/mongodb-ssl/certs/ directory.
-    docker pull mongo:3.4.4
+    docker pull mongo:3.4
     docker run -d \
         --name mdb-with-ssl \
         --publish=27017:27017 \
         --volume=${TRAVIS_BUILD_DIR}/tests/backend/mongodb-ssl/certs:/certs \
-        mongo:3.4.4 \
-        --replSet=bigchain-rs \
+        mongo:3.4 \
+        # --replSet=bigchain-rs \
         --sslAllowInvalidHostnames \
         --sslMode=requireSSL \
         --sslCAFile=/certs/ca.crt \

.ci/travis_script.sh

@@ -4,14 +4,14 @@ set -e -x
 
 if [[ -n ${TOXENV} ]]; then
   tox -e ${TOXENV}
-elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == mongodb && \
+elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == localmongodb && \
     -z "${BIGCHAINDB_DATABASE_SSL}" ]]; then
     # Run the full suite of tests for MongoDB over an unsecure connection
-  pytest -sv --database-backend=mongodb --cov=bigchaindb
-elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == mongodb && \
+  pytest -sv --database-backend=localmongodb --cov=bigchaindb
+elif [[ "${BIGCHAINDB_DATABASE_BACKEND}" == localmongodb && \
     "${BIGCHAINDB_DATABASE_SSL}" == true ]]; then
     # Run a sub-set of tests over SSL; those marked as 'pytest.mark.bdb_ssl'.
-  pytest -sv --database-backend=mongodb-ssl --cov=bigchaindb -m bdb_ssl
+  pytest -sv --database-backend=localmongodb-ssl --cov=bigchaindb -m bdb_ssl
 else
   pytest -sv -n auto --cov=bigchaindb
 fi

.travis.yml

@@ -9,9 +9,8 @@ language: python
 cache: pip
 
 python:
-  - 3.5
   - 3.6
-  
+
 env:
   - TOXENV=flake8
   - TOXENV=docsroot
@@ -19,34 +18,11 @@ env:
 
 matrix:
   fast_finish: true
-  exclude:
-  - python: 3.5
-    env: TOXENV=flake8 
-  - python: 3.5 
-    env: TOXENV=docsroot
-  - python: 3.5 
-    env: TOXENV=docsserver
   include:
-  - python: 3.5
-    env: BIGCHAINDB_DATABASE_BACKEND=rethinkdb
-  - python: 3.5
+  - python: 3.6
     env:
-    - BIGCHAINDB_DATABASE_BACKEND=mongodb
+    - BIGCHAINDB_DATABASE_BACKEND=localmongodb
     - BIGCHAINDB_DATABASE_SSL=
-  - python: 3.6
-    env: BIGCHAINDB_DATABASE_BACKEND=rethinkdb
-  - python: 3.6
-    env:
-    - BIGCHAINDB_DATABASE_BACKEND=mongodb
-    - BIGCHAINDB_DATABASE_SSL=
-  - python: 3.5
-    env:
-    - BIGCHAINDB_DATABASE_BACKEND=mongodb
-    - BIGCHAINDB_DATABASE_SSL=true
-  - python: 3.6
-    env:
-    - BIGCHAINDB_DATABASE_BACKEND=mongodb
-    - BIGCHAINDB_DATABASE_SSL=true
 
 before_install: sudo .ci/travis-before-install.sh
 

Makefile

@@ -57,7 +57,7 @@ test-all: ## run tests on every Python version with tox
 	tox
 
 coverage: ## check code coverage quickly with the default Python
-	pytest -v -n auto --cov=bigchaindb --cov-report term --cov-report html
+	pytest -v -n auto --database-backend=localmongodb --cov=bigchaindb --cov-report term --cov-report html
 	$(BROWSER) htmlcov/index.html
 
 docs: ## generate Sphinx HTML documentation, including API docs

bigchaindb/__init__.py

@@ -21,10 +21,20 @@ _base_database_rethinkdb = {
 # because dicts are unordered. I tried to configure
 
 _database_keys_map = {
+    'localmongodb': ('host', 'port', 'name'),
     'mongodb': ('host', 'port', 'name', 'replicaset'),
     'rethinkdb': ('host', 'port', 'name')
 }
 
+_base_database_localmongodb = {
+    'host': os.environ.get('BIGCHAINDB_DATABASE_HOST', 'localhost'),
+    'port': int(os.environ.get('BIGCHAINDB_DATABASE_PORT', 27017)),
+    'name': os.environ.get('BIGCHAINDB_DATABASE_NAME', 'bigchain'),
+    'replicaset': os.environ.get('BIGCHAINDB_DATABASE_REPLICASET'),
+    'login': os.environ.get('BIGCHAINDB_DATABASE_LOGIN'),
+    'password': os.environ.get('BIGCHAINDB_DATABASE_PASSWORD')
+}
+
 _base_database_mongodb = {
     'host': os.environ.get('BIGCHAINDB_DATABASE_HOST', 'localhost'),
     'port': int(os.environ.get('BIGCHAINDB_DATABASE_PORT', 27017)),
@@ -54,7 +64,21 @@ _database_mongodb = {
 }
 _database_mongodb.update(_base_database_mongodb)
 
+_database_localmongodb = {
+    'backend': os.environ.get('BIGCHAINDB_DATABASE_BACKEND', 'localmongodb'),
+    'connection_timeout': 5000,
+    'max_tries': 3,
+    'ssl': bool(os.environ.get('BIGCHAINDB_DATABASE_SSL', False)),
+    'ca_cert': os.environ.get('BIGCHAINDB_DATABASE_CA_CERT'),
+    'certfile': os.environ.get('BIGCHAINDB_DATABASE_CERTFILE'),
+    'keyfile': os.environ.get('BIGCHAINDB_DATABASE_KEYFILE'),
+    'keyfile_passphrase': os.environ.get('BIGCHAINDB_DATABASE_KEYFILE_PASSPHRASE'),
+    'crlfile': os.environ.get('BIGCHAINDB_DATABASE_CRLFILE')
+}
+_database_localmongodb.update(_base_database_localmongodb)
+
 _database_map = {
+    'localmongodb': _database_localmongodb,
     'mongodb': _database_mongodb,
     'rethinkdb': _database_rethinkdb
 }

bigchaindb/backend/connection.py

@@ -8,6 +8,7 @@ from bigchaindb.backend.exceptions import ConnectionError
 
 
 BACKENDS = {
+    'localmongodb': 'bigchaindb.backend.localmongodb.connection.LocalMongoDBConnection',
     'mongodb': 'bigchaindb.backend.mongodb.connection.MongoDBConnection',
     'rethinkdb': 'bigchaindb.backend.rethinkdb.connection.RethinkDBConnection'
 }

bigchaindb/backend/localmongodb/__init__.py

@@ -0,0 +1,22 @@
+"""MongoDB backend implementation.
+
+Contains a MongoDB-specific implementation of the
+:mod:`~bigchaindb.backend.changefeed`, :mod:`~bigchaindb.backend.query`, and
+:mod:`~bigchaindb.backend.schema` interfaces.
+
+You can specify BigchainDB to use MongoDB as its database backend by either
+setting ``database.backend`` to ``'rethinkdb'`` in your configuration file, or
+setting the ``BIGCHAINDB_DATABASE_BACKEND`` environment variable to
+``'rethinkdb'``.
+
+If configured to use MongoDB, BigchainDB will automatically return instances
+of :class:`~bigchaindb.backend.rethinkdb.MongoDBConnection` for
+:func:`~bigchaindb.backend.connection.connect` and dispatch calls of the
+generic backend interfaces to the implementations in this module.
+"""
+
+# Register the single dispatched modules on import.
+from bigchaindb.backend.localmongodb import schema, query # noqa
+
+# MongoDBConnection should always be accessed via
+# ``bigchaindb.backend.connect()``.

bigchaindb/backend/localmongodb/connection.py

@@ -0,0 +1,5 @@
+from bigchaindb.backend.mongodb.connection import MongoDBConnection
+
+
+class LocalMongoDBConnection(MongoDBConnection):
+    pass

bigchaindb/backend/localmongodb/query.py

@@ -0,0 +1,41 @@
+"""Query implementation for MongoDB"""
+
+from bigchaindb import backend
+from bigchaindb.backend.exceptions import DuplicateKeyError
+from bigchaindb.backend.utils import module_dispatch_registrar
+from bigchaindb.backend.localmongodb.connection import LocalMongoDBConnection
+
+
+register_query = module_dispatch_registrar(backend.query)
+
+
+@register_query(LocalMongoDBConnection)
+def write_transaction(conn, signed_transaction):
+    try:
+        return conn.run(
+            conn.collection('transactions')
+            .insert_one(signed_transaction))
+    except DuplicateKeyError:
+        pass
+
+
+@register_query(LocalMongoDBConnection)
+def get_transaction(conn, transaction_id):
+    try:
+        return conn.run(
+            conn.collection('transactions')
+            .find_one({'id': transaction_id}, {'_id': 0}))
+    except IndexError:
+        pass
+
+
+@register_query(LocalMongoDBConnection)
+def get_spent(conn, transaction_id, output):
+    try:
+        return conn.run(
+            conn.collection('transactions')
+            .find_one({'id': transaction_id,
+                       'inputs.fulfills.output_index': output},
+                      {'_id': 0}))
+    except IndexError:
+        pass

bigchaindb/backend/localmongodb/schema.py

@@ -0,0 +1,82 @@
+"""Utils to initialize and drop the database."""
+
+import logging
+
+from pymongo import ASCENDING, TEXT
+
+from bigchaindb import backend
+from bigchaindb.common import exceptions
+from bigchaindb.backend.utils import module_dispatch_registrar
+from bigchaindb.backend.localmongodb.connection import LocalMongoDBConnection
+
+
+logger = logging.getLogger(__name__)
+register_schema = module_dispatch_registrar(backend.schema)
+
+
+@register_schema(LocalMongoDBConnection)
+def create_database(conn, dbname):
+    if dbname in conn.conn.database_names():
+        raise exceptions.DatabaseAlreadyExists('Database `{}` already exists'
+                                               .format(dbname))
+
+    logger.info('Create database `%s`.', dbname)
+    # TODO: read and write concerns can be declared here
+    conn.conn.get_database(dbname)
+
+
+@register_schema(LocalMongoDBConnection)
+def create_tables(conn, dbname):
+    for table_name in ['transactions', 'assets']:
+        logger.info('Create `%s` table.', table_name)
+        # create the table
+        # TODO: read and write concerns can be declared here
+        conn.conn[dbname].create_collection(table_name)
+
+
+@register_schema(LocalMongoDBConnection)
+def create_indexes(conn, dbname):
+    create_transactions_secondary_index(conn, dbname)
+    create_assets_secondary_index(conn, dbname)
+
+
+@register_schema(LocalMongoDBConnection)
+def drop_database(conn, dbname):
+    conn.conn.drop_database(dbname)
+
+
+def create_transactions_secondary_index(conn, dbname):
+    logger.info('Create `transactions` secondary index.')
+
+    # to query the transactions for a transaction id, this field is unique
+    conn.conn[dbname]['transactions'].create_index('transactions.id',
+                                                   name='transaction_id')
+
+    # secondary index for asset uuid, this field is unique
+    conn.conn[dbname]['transactions']\
+        .create_index('asset.id', name='asset_id')
+
+    # secondary index on the public keys of outputs
+    conn.conn[dbname]['transactions']\
+        .create_index('outputs.public_keys',
+                      name='outputs')
+
+    # secondary index on inputs/transaction links (transaction_id, output)
+    conn.conn[dbname]['transactions']\
+        .create_index([
+            ('inputs.fulfills.transaction_id', ASCENDING),
+            ('inputs.fulfills.output_index', ASCENDING),
+        ], name='inputs')
+
+
+def create_assets_secondary_index(conn, dbname):
+    logger.info('Create `assets` secondary index.')
+
+    # unique index on the id of the asset.
+    # the id is the txid of the transaction that created the asset
+    conn.conn[dbname]['assets'].create_index('id',
+                                             name='asset_id',
+                                             unique=True)
+
+    # full text search index
+    conn.conn[dbname]['assets'].create_index([('$**', TEXT)], name='text')

bigchaindb/backend/mongodb/connection.py

@@ -84,21 +84,22 @@ class MongoDBConnection(Connection):
         """
 
         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,
-                                   self.dbname,
-                                   self.ssl,
-                                   self.login,
-                                   self.password,
-                                   self.ca_cert,
-                                   self.certfile,
-                                   self.keyfile,
-                                   self.keyfile_passphrase,
-                                   self.crlfile)
+            if self.replicaset:
+                # 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,
+                                       self.dbname,
+                                       self.ssl,
+                                       self.login,
+                                       self.password,
+                                       self.ca_cert,
+                                       self.certfile,
+                                       self.keyfile,
+                                       self.keyfile_passphrase,
+                                       self.crlfile)
 
             # FYI: the connection process might raise a
             # `ServerSelectionTimeoutError`, that is a subclass of

bigchaindb/backend/query.py

@@ -19,6 +19,20 @@ def write_transaction(connection, signed_transaction):
     raise NotImplementedError
 
 
+@singledispatch
+def get_transaction(connection, signed_transaction):
+    """Get a transaction from the transactions table.
+
+    Args:
+        signed_transaction (dict): a signed transaction.
+
+    Returns:
+        The result of the operation.
+    """
+
+    raise NotImplementedError
+
+
 @singledispatch
 def update_transaction(connection, transaction_id, doc):
     """Update a transaction in the backlog table.

bigchaindb/commands/bigchaindb.py

@@ -270,7 +270,7 @@ def create_parser():
                                           help='Prepare the config file '
                                                'and create the node keypair')
     config_parser.add_argument('backend',
-                               choices=['rethinkdb', 'mongodb'],
+                               choices=['rethinkdb', 'mongodb', 'localmongodb'],
                                help='The backend to use. It can be either '
                                     'rethinkdb or mongodb.')
 

bigchaindb/models.py

@@ -30,7 +30,11 @@ class Transaction(Transaction):
         """
         input_conditions = []
 
-        if self.operation == Transaction.TRANSFER:
+        if self.operation == Transaction.CREATE:
+            if bigchain.get_transaction(self.to_dict()['id']):
+                raise DuplicateTransaction('transaction `{}` already exists'
+                                           .format(self.id))
+        elif self.operation == Transaction.TRANSFER:
             # store the inputs so that we can check if the asset ids match
             input_txs = []
             for input_ in self.inputs:

bigchaindb/tendermint/__init__.py

@@ -0,0 +1,5 @@
+# Order is important!
+# If we import core first, core will try to load BigchainDB from
+# __init__ itself, causing a loop.
+from bigchaindb.tendermint.lib import BigchainDB  # noqa
+from bigchaindb.tendermint.core import App  # noqa

bigchaindb/tendermint/commands.py

@@ -0,0 +1,12 @@
+from abci import ABCIServer
+
+from bigchaindb.tendermint.core import App
+
+
+def start():
+    app = ABCIServer(app=App())
+    app.run()
+
+
+if __name__ == '__main__':
+    start()

bigchaindb/tendermint/core.py

@@ -0,0 +1,49 @@
+"""This module contains all the goodness to integrate BigchainDB
+with Tendermint."""
+
+
+from abci import BaseApplication, Result
+
+from bigchaindb.tendermint import BigchainDB
+from bigchaindb.tendermint.utils import decode_transaction
+
+
+class App(BaseApplication):
+    """Bridge between BigchainDB and Tendermint.
+
+    The role of this class is to expose the BigchainDB
+    transactional logic to the Tendermint Consensus
+    State Machine."""
+
+    def __init__(self, bigchaindb=None):
+        if not bigchaindb:
+            bigchaindb = BigchainDB()
+        self.bigchaindb = bigchaindb
+
+    def check_tx(self, raw_transaction):
+        """Validate the transaction before entry into
+        the mempool.
+
+        Args:
+            raw_tx: an encoded transaction."""
+
+        transaction = decode_transaction(raw_transaction)
+        if self.bigchaindb.validate_transaction(transaction):
+            return Result.ok()
+        else:
+            return Result.error()
+
+    def deliver_tx(self, raw_transaction):
+        """Validate the transaction before mutating the state.
+
+        Args:
+            raw_tx: an encoded transaction."""
+
+        transaction = self.bigchaindb.validate_transaction(
+                decode_transaction(raw_transaction))
+
+        if not transaction:
+            return Result.error(log='Invalid transaction')
+        else:
+            self.bigchaindb.write_transaction(transaction)
+            return Result.ok()

bigchaindb/tendermint/lib.py

@@ -0,0 +1,52 @@
+import logging
+
+from bigchaindb import backend
+from bigchaindb import Bigchain
+from bigchaindb.models import Transaction
+from bigchaindb.common.exceptions import SchemaValidationError, ValidationError
+
+
+logger = logging.getLogger(__name__)
+
+
+class BigchainDB(Bigchain):
+
+    def write_transaction(self, transaction):
+        """Write a valid transaction to the transactions collection."""
+
+        return backend.query.write_transaction(self.connection, transaction.to_dict())
+
+    def get_transaction(self, transaction, include_status=False):
+        result = backend.query.get_transaction(self.connection, transaction)
+
+        if result:
+            result = Transaction.from_dict(result)
+
+        if include_status:
+            return result, self.TX_VALID if result else None
+        else:
+            return result
+
+    def get_spent(self, txid, output):
+        transaction = backend.query.get_spent(self.connection, txid, output)
+        return Transaction.from_dict(transaction)
+
+    def validate_transaction(self, tx):
+        """Validate a transaction against the current status
+        of the database."""
+
+        try:
+            tx_obj = Transaction.from_dict(tx)
+        except SchemaValidationError as e:
+            logger.warning('Invalid transaction schema: %s', e.__cause__.message)
+            return False
+        except ValidationError as e:
+            logger.warning('Invalid transaction (%s): %s', type(e).__name__, e)
+            return False
+
+        try:
+            return tx_obj.validate(self)
+        except ValidationError as e:
+            logger.warning('Invalid transaction (%s): %s', type(e).__name__, e)
+            return False
+        return tx_obj

bigchaindb/tendermint/utils.py

@@ -0,0 +1,14 @@
+import base64
+import json
+
+
+def encode_transaction(value):
+    """Encode a transaction to Base64."""
+
+    return base64.b64encode(json.dumps(value).encode('utf8'))
+
+
+def decode_transaction(raw):
+    """Decode a transaction from Base64 to a dict."""
+
+    return json.loads(base64.b64decode(raw.decode('utf8')))

pytest.ini

@@ -1,3 +1,3 @@
 [pytest]
-testpaths = tests
+testpaths = tests/tendermint
 norecursedirs = .* *.egg *.egg-info env* devenv* docs

setup.py

@@ -83,6 +83,7 @@ install_requires = [
     'aiohttp~=2.0',
     'python-rapidjson-schema==0.1.1',
     'statsd==3.2.1',
+    'abci~=0.2.0',
 ]
 
 setup(

tests/tendermint/conftest.py

@@ -0,0 +1,7 @@
+import pytest
+
+
+@pytest.fixture
+def b():
+    from bigchaindb.tendermint import BigchainDB
+    return BigchainDB()

tests/tendermint/test_core.py

@@ -0,0 +1,73 @@
+def test_check_tx__signed_create_is_ok(b):
+    from bigchaindb.tendermint import App
+    from bigchaindb.tendermint.utils import encode_transaction
+    from bigchaindb.models import Transaction
+    from bigchaindb.common.crypto import generate_key_pair
+
+    alice = generate_key_pair()
+    bob = generate_key_pair()
+
+    tx = Transaction.create([alice.public_key],
+                            [([bob.public_key], 1)])\
+                    .sign([alice.private_key])
+
+    app = App(b)
+    result = app.check_tx(encode_transaction(tx.to_dict()))
+    assert result.is_ok()
+
+
+def test_check_tx__unsigned_create_is_error(b):
+    from bigchaindb.tendermint import App
+    from bigchaindb.tendermint.utils import encode_transaction
+    from bigchaindb.models import Transaction
+    from bigchaindb.common.crypto import generate_key_pair
+
+    alice = generate_key_pair()
+    bob = generate_key_pair()
+
+    tx = Transaction.create([alice.public_key],
+                            [([bob.public_key], 1)])
+
+    app = App(b)
+    result = app.check_tx(encode_transaction(tx.to_dict()))
+    assert result.is_error()
+
+
+def test_deliver_tx__valid_create_updates_db(b):
+    from bigchaindb.tendermint import App
+    from bigchaindb.tendermint.utils import encode_transaction
+    from bigchaindb.models import Transaction
+    from bigchaindb.common.crypto import generate_key_pair
+
+    alice = generate_key_pair()
+    bob = generate_key_pair()
+
+    tx = Transaction.create([alice.public_key],
+                            [([bob.public_key], 1)])\
+                    .sign([alice.private_key])
+
+    app = App(b)
+    result = app.deliver_tx(encode_transaction(tx.to_dict()))
+    assert result.is_ok()
+    assert b.get_transaction(tx.id).id == tx.id
+
+
+def test_deliver_tx__double_spend_fails(b):
+    from bigchaindb.tendermint import App
+    from bigchaindb.tendermint.utils import encode_transaction
+    from bigchaindb.models import Transaction
+    from bigchaindb.common.crypto import generate_key_pair
+
+    alice = generate_key_pair()
+    bob = generate_key_pair()
+
+    tx = Transaction.create([alice.public_key],
+                            [([bob.public_key], 1)])\
+                    .sign([alice.private_key])
+
+    app = App(b)
+    result = app.deliver_tx(encode_transaction(tx.to_dict()))
+    assert result.is_ok()
+    assert b.get_transaction(tx.id).id == tx.id
+    result = app.deliver_tx(encode_transaction(tx.to_dict()))
+    assert result.is_error()