Implement UTXO set backend (#2033)

bigchaindb/backend/localmongodb/query.py

@@ -199,3 +199,36 @@ def get_block_with_transaction(conn, txid):
         conn.collection('blocks')
         .find({'transactions': txid},
               projection={'_id': False, 'height': True}))
+
+
+@register_query(LocalMongoDBConnection)
+def store_unspent_outputs(conn, *unspent_outputs):
+    try:
+        return conn.run(
+            conn.collection('utxos')
+            .insert_many(unspent_outputs, ordered=False))
+    except DuplicateKeyError:
+        # TODO log warning at least
+        pass
+
+
+@register_query(LocalMongoDBConnection)
+def delete_unspent_outputs(conn, *unspent_outputs):
+    cursor = conn.run(
+            conn.collection('utxos').remove(
+                {'$or': [
+                    {'$and': [
+                        {'transaction_id': unspent_output['transaction_id']},
+                        {'output_index': unspent_output['output_index']}
+                    ]}
+                    for unspent_output in unspent_outputs
+                ]}
+                ))
+    return cursor
+
+
+@register_query(LocalMongoDBConnection)
+def get_unspent_outputs(conn, *, query=None):
+    if query is None:
+        query = {}
+    return conn.run(conn.collection('utxos').find(query))

bigchaindb/backend/localmongodb/schema.py

@@ -27,7 +27,7 @@ def create_database(conn, dbname):
 
 @register_schema(LocalMongoDBConnection)
 def create_tables(conn, dbname):
-    for table_name in ['transactions', 'assets', 'blocks', 'metadata']:
+    for table_name in ['transactions', 'utxos', 'assets', 'blocks', 'metadata']:
         logger.info('Create `%s` table.', table_name)
         # create the table
         # TODO: read and write concerns can be declared here
@@ -40,6 +40,7 @@ def create_indexes(conn, dbname):
     create_assets_secondary_index(conn, dbname)
     create_blocks_secondary_index(conn, dbname)
     create_metadata_secondary_index(conn, dbname)
+    create_utxos_secondary_index(conn, dbname)
 
 
 @register_schema(LocalMongoDBConnection)
@@ -99,3 +100,13 @@ def create_metadata_secondary_index(conn, dbname):
 
     # full text search index
     conn.conn[dbname]['metadata'].create_index([('$**', TEXT)], name='text')
+
+
+def create_utxos_secondary_index(conn, dbname):
+    logger.info('Create `utxos` secondary index.')
+
+    conn.conn[dbname]['utxos'].create_index(
+        [('transaction_id', ASCENDING), ('output_index', ASCENDING)],
+        name='utxo',
+        unique=True,
+    )

bigchaindb/backend/query.py

@@ -547,3 +547,34 @@ def store_block(conn, block):
     """
 
     raise NotImplementedError
+
+
+@singledispatch
+def store_unspent_outputs(connection, unspent_outputs):
+    """Store unspent outputs in ``utxo_set`` table."""
+
+    raise NotImplementedError
+
+
+@singledispatch
+def delete_unspent_outputs(connection, unspent_outputs):
+    """Delete unspent outputs in ``utxo_set`` table."""
+
+    raise NotImplementedError
+
+
+@singledispatch
+def get_unspent_outputs(connection, *, query=None):
+    """Retrieves unspent outputs.
+
+    Args:
+        query (dict): An optional parameter to filter the result set.
+            Defaults to ``None``, which means that all UTXO records
+            will be returned.
+
+    Returns:
+        Generator yielding unspent outputs (UTXO set) according to the
+        given query.
+    """
+
+    raise NotImplementedError

tests/backend/localmongodb/conftest.py

@@ -0,0 +1,12 @@
+from pymongo import MongoClient
+from pytest import fixture
+
+
+@fixture
+def mongo_client(db_context):
+    return MongoClient(host=db_context.host, port=db_context.port)
+
+
+@fixture
+def utxo_collection(db_context, mongo_client):
+    return mongo_client[db_context.name].utxos

tests/backend/localmongodb/test_queries.py

@@ -6,6 +6,23 @@ import pymongo
 pytestmark = [pytest.mark.tendermint, pytest.mark.localmongodb, pytest.mark.bdb]
 
 
+@pytest.fixture
+def dummy_unspent_outputs():
+    return [
+        {'transaction_id': 'a', 'output_index': 0},
+        {'transaction_id': 'a', 'output_index': 1},
+        {'transaction_id': 'b', 'output_index': 0},
+    ]
+
+
+@pytest.fixture
+def utxoset(dummy_unspent_outputs, utxo_collection):
+    insert_res = utxo_collection.insert_many(deepcopy(dummy_unspent_outputs))
+    assert insert_res.acknowledged
+    assert len(insert_res.inserted_ids) == 3
+    return dummy_unspent_outputs, utxo_collection
+
+
 def test_get_txids_filtered(signed_create_tx, signed_transfer_tx):
     from bigchaindb.backend import connect, query
     from bigchaindb.models import Transaction
@@ -180,3 +197,54 @@ def test_get_block():
 
     block = dict(query.get_block(conn, 3))
     assert block['height'] == 3
+
+
+def test_delete_unspent_outputs(db_context, utxoset):
+    from bigchaindb.backend import query
+    unspent_outputs, utxo_collection = utxoset
+    delete_res = query.delete_unspent_outputs(db_context.conn,
+                                              *unspent_outputs[::2])
+    assert delete_res['n'] == 2
+    assert utxo_collection.find(
+        {'$or': [
+            {'transaction_id': 'a', 'output_index': 0},
+            {'transaction_id': 'b', 'output_index': 0},
+        ]}
+    ).count() == 0
+    assert utxo_collection.find(
+            {'transaction_id': 'a', 'output_index': 1}).count() == 1
+
+
+def test_store_one_unspent_output(db_context,
+                                  unspent_output_1, utxo_collection):
+    from bigchaindb.backend import query
+    res = query.store_unspent_outputs(db_context.conn, unspent_output_1)
+    assert res.acknowledged
+    assert len(res.inserted_ids) == 1
+    assert utxo_collection.find(
+        {'transaction_id': unspent_output_1['transaction_id'],
+         'output_index': unspent_output_1['output_index']}
+    ).count() == 1
+
+
+def test_store_many_unspent_outputs(db_context,
+                                    unspent_outputs, utxo_collection):
+    from bigchaindb.backend import query
+    res = query.store_unspent_outputs(db_context.conn, *unspent_outputs)
+    assert res.acknowledged
+    assert len(res.inserted_ids) == 3
+    assert utxo_collection.find(
+        {'transaction_id': unspent_outputs[0]['transaction_id']}
+    ).count() == 3
+
+
+def test_get_unspent_outputs(db_context, utxoset):
+    from bigchaindb.backend import query
+    cursor = query.get_unspent_outputs(db_context.conn)
+    assert cursor.count() == 3
+    retrieved_utxoset = list(cursor)
+    unspent_outputs, utxo_collection = utxoset
+    assert retrieved_utxoset == list(utxo_collection.find())
+    for utxo in retrieved_utxoset:
+        del utxo['_id']
+    assert retrieved_utxoset == unspent_outputs

tests/backend/localmongodb/test_schema.py

@@ -0,0 +1,110 @@
+import pytest
+
+
+pytestmark = [pytest.mark.bdb, pytest.mark.tendermint]
+
+
+def test_init_creates_db_tables_and_indexes():
+    import bigchaindb
+    from bigchaindb import backend
+    from bigchaindb.backend.schema import init_database
+
+    conn = backend.connect()
+    dbname = bigchaindb.config['database']['name']
+
+    # the db is set up by the fixture so we need to remove it
+    conn.conn.drop_database(dbname)
+
+    init_database()
+
+    collection_names = conn.conn[dbname].collection_names()
+    assert set(collection_names) == {
+            'transactions', 'assets', 'metadata', 'blocks', 'utxos'}
+
+    indexes = conn.conn[dbname]['assets'].index_information().keys()
+    assert set(indexes) == {'_id_', 'asset_id', 'text'}
+
+    indexes = conn.conn[dbname]['transactions'].index_information().keys()
+    assert set(indexes) == {
+            '_id_', 'transaction_id', 'asset_id', 'outputs', 'inputs'}
+
+    indexes = conn.conn[dbname]['blocks'].index_information().keys()
+    assert set(indexes) == {'_id_', 'height'}
+
+    indexes = conn.conn[dbname]['utxos'].index_information().keys()
+    assert set(indexes) == {'_id_', 'utxo'}
+
+
+def test_init_database_fails_if_db_exists():
+    import bigchaindb
+    from bigchaindb import backend
+    from bigchaindb.backend.schema import init_database
+    from bigchaindb.common import exceptions
+
+    conn = backend.connect()
+    dbname = bigchaindb.config['database']['name']
+
+    # The db is set up by the fixtures
+    assert dbname in conn.conn.database_names()
+
+    with pytest.raises(exceptions.DatabaseAlreadyExists):
+        init_database()
+
+
+def test_create_tables():
+    import bigchaindb
+    from bigchaindb import backend
+    from bigchaindb.backend import schema
+
+    conn = backend.connect()
+    dbname = bigchaindb.config['database']['name']
+
+    # The db is set up by the fixtures so we need to remove it
+    conn.conn.drop_database(dbname)
+    schema.create_database(conn, dbname)
+    schema.create_tables(conn, dbname)
+
+    collection_names = conn.conn[dbname].collection_names()
+    assert set(collection_names) == {
+            'transactions', 'assets', 'metadata', 'blocks', 'utxos'}
+
+
+def test_create_secondary_indexes():
+    import bigchaindb
+    from bigchaindb import backend
+    from bigchaindb.backend import schema
+
+    conn = backend.connect()
+    dbname = bigchaindb.config['database']['name']
+
+    # The db is set up by the fixtures so we need to remove it
+    conn.conn.drop_database(dbname)
+    schema.create_database(conn, dbname)
+    schema.create_tables(conn, dbname)
+    schema.create_indexes(conn, dbname)
+
+    indexes = conn.conn[dbname]['assets'].index_information().keys()
+    assert set(indexes) == {'_id_', 'asset_id', 'text'}
+
+    indexes = conn.conn[dbname]['transactions'].index_information().keys()
+    assert set(indexes) == {
+            '_id_', 'transaction_id', 'asset_id', 'outputs', 'inputs'}
+
+    indexes = conn.conn[dbname]['blocks'].index_information().keys()
+    assert set(indexes) == {'_id_', 'height'}
+
+    index_info = conn.conn[dbname]['utxos'].index_information()
+    assert set(index_info.keys()) == {'_id_', 'utxo'}
+    assert index_info['utxo']['unique']
+    assert index_info['utxo']['key'] == [('transaction_id', 1),
+                                         ('output_index', 1)]
+
+
+def test_drop(dummy_db):
+    from bigchaindb import backend
+    from bigchaindb.backend import schema
+
+    conn = backend.connect()
+    assert dummy_db in conn.conn.database_names()
+    schema.drop_database(conn, dummy_db)
+    assert dummy_db not in conn.conn.database_names()

tests/conftest.py

@@ -617,3 +617,44 @@ def genesis_tx(b, user_pk):
     tx.operation = Transaction.GENESIS
     genesis_tx = tx.sign([b.me_private])
     return genesis_tx
+
+
+@pytest.fixture
+def unspent_output_0():
+    return {
+        'amount': 1,
+        'asset_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d',
+        'condition_uri': 'ni:///sha-256;RmovleG60-7K0CX60jjfUunV3lBpUOkiQOAnBzghm0w?fpt=ed25519-sha-256&cost=131072',
+        'fulfillment_message': '{"asset":{"data":{"hash":"06e47bcf9084f7ecfd2a2a2ad275444a"}},"id":"e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d","inputs":[{"fulfillment":"pGSAIIQT0Jm6LDlcSs9coJK4Q4W-SNtsO2EtMtQJ04EUjBMJgUAXKIqeaippbF-IClhhZNNaP6EIZ_OgrVQYU4mH6b-Vc3Tg-k6p-rJOlLGUUo_w8C5QgPHNRYFOqUk2f1q0Cs4G","fulfills":null,"owners_before":["9taLkHkaBXeSF8vrhDGFTAmcZuCEPqjQrKadfYGs4gHv"]}],"metadata":null,"operation":"CREATE","outputs":[{"amount":"1","condition":{"details":{"public_key":"6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz","type":"ed25519-sha-256"},"uri":"ni:///sha-256;RmovleG60-7K0CX60jjfUunV3lBpUOkiQOAnBzghm0w?fpt=ed25519-sha-256&cost=131072"},"public_keys":["6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz"]},{"amount":"2","condition":{"details":{"public_key":"AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT","type":"ed25519-sha-256"},"uri":"ni:///sha-256;-HlYmgwwl-vXwE52IaADhvYxaL1TbjqfJ-LGn5a1PFc?fpt=ed25519-sha-256&cost=131072"},"public_keys":["AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT"]},{"amount":"3","condition":{"details":{"public_key":"HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB","type":"ed25519-sha-256"},"uri":"ni:///sha-256;xfn8pvQkTCPtvR0trpHy2pqkkNTmMBCjWMMOHtk3WO4?fpt=ed25519-sha-256&cost=131072"},"public_keys":["HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB"]}],"version":"1.0"}',   # noqa
+        'output_index': 0,
+        'transaction_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d'
+    }
+
+
+@pytest.fixture
+def unspent_output_1():
+    return {
+        'amount': 2,
+        'asset_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d',
+        'condition_uri': 'ni:///sha-256;-HlYmgwwl-vXwE52IaADhvYxaL1TbjqfJ-LGn5a1PFc?fpt=ed25519-sha-256&cost=131072',
+        'fulfillment_message': '{"asset":{"data":{"hash":"06e47bcf9084f7ecfd2a2a2ad275444a"}},"id":"e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d","inputs":[{"fulfillment":"pGSAIIQT0Jm6LDlcSs9coJK4Q4W-SNtsO2EtMtQJ04EUjBMJgUAXKIqeaippbF-IClhhZNNaP6EIZ_OgrVQYU4mH6b-Vc3Tg-k6p-rJOlLGUUo_w8C5QgPHNRYFOqUk2f1q0Cs4G","fulfills":null,"owners_before":["9taLkHkaBXeSF8vrhDGFTAmcZuCEPqjQrKadfYGs4gHv"]}],"metadata":null,"operation":"CREATE","outputs":[{"amount":"1","condition":{"details":{"public_key":"6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz","type":"ed25519-sha-256"},"uri":"ni:///sha-256;RmovleG60-7K0CX60jjfUunV3lBpUOkiQOAnBzghm0w?fpt=ed25519-sha-256&cost=131072"},"public_keys":["6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz"]},{"amount":"2","condition":{"details":{"public_key":"AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT","type":"ed25519-sha-256"},"uri":"ni:///sha-256;-HlYmgwwl-vXwE52IaADhvYxaL1TbjqfJ-LGn5a1PFc?fpt=ed25519-sha-256&cost=131072"},"public_keys":["AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT"]},{"amount":"3","condition":{"details":{"public_key":"HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB","type":"ed25519-sha-256"},"uri":"ni:///sha-256;xfn8pvQkTCPtvR0trpHy2pqkkNTmMBCjWMMOHtk3WO4?fpt=ed25519-sha-256&cost=131072"},"public_keys":["HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB"]}],"version":"1.0"}',   # noqa
+        'output_index': 1,
+        'transaction_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d',
+    }
+
+
+@pytest.fixture
+def unspent_output_2():
+    return {
+        'amount': 3,
+        'asset_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d',
+        'condition_uri': 'ni:///sha-256;xfn8pvQkTCPtvR0trpHy2pqkkNTmMBCjWMMOHtk3WO4?fpt=ed25519-sha-256&cost=131072',
+        'fulfillment_message': '{"asset":{"data":{"hash":"06e47bcf9084f7ecfd2a2a2ad275444a"}},"id":"e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d","inputs":[{"fulfillment":"pGSAIIQT0Jm6LDlcSs9coJK4Q4W-SNtsO2EtMtQJ04EUjBMJgUAXKIqeaippbF-IClhhZNNaP6EIZ_OgrVQYU4mH6b-Vc3Tg-k6p-rJOlLGUUo_w8C5QgPHNRYFOqUk2f1q0Cs4G","fulfills":null,"owners_before":["9taLkHkaBXeSF8vrhDGFTAmcZuCEPqjQrKadfYGs4gHv"]}],"metadata":null,"operation":"CREATE","outputs":[{"amount":"1","condition":{"details":{"public_key":"6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz","type":"ed25519-sha-256"},"uri":"ni:///sha-256;RmovleG60-7K0CX60jjfUunV3lBpUOkiQOAnBzghm0w?fpt=ed25519-sha-256&cost=131072"},"public_keys":["6FDGsHrR9RZqNaEm7kBvqtxRkrvuWogBW2Uy7BkWc5Tz"]},{"amount":"2","condition":{"details":{"public_key":"AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT","type":"ed25519-sha-256"},"uri":"ni:///sha-256;-HlYmgwwl-vXwE52IaADhvYxaL1TbjqfJ-LGn5a1PFc?fpt=ed25519-sha-256&cost=131072"},"public_keys":["AH9D7xgmhyLmVE944zvHvuvYWuj5DfbMBJhnDM4A5FdT"]},{"amount":"3","condition":{"details":{"public_key":"HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB","type":"ed25519-sha-256"},"uri":"ni:///sha-256;xfn8pvQkTCPtvR0trpHy2pqkkNTmMBCjWMMOHtk3WO4?fpt=ed25519-sha-256&cost=131072"},"public_keys":["HpmSVrojHvfCXQbmoAs4v6Aq1oZiZsZDnjr68KiVtPbB"]}],"version":"1.0"}',   # noqa
+        'output_index': 2,
+        'transaction_id': 'e897c7a0426461a02b4fca8ed73bc0debed7570cf3b40fb4f49c963434225a4d',
+    }
+
+
+@pytest.fixture
+def unspent_outputs(unspent_output_0, unspent_output_1, unspent_output_2):
+    return unspent_output_0, unspent_output_1, unspent_output_2

tests/utils.py

@@ -46,6 +46,7 @@ def flush_mongo_db(connection, dbname):
     connection.conn[dbname].votes.delete_many({})
     connection.conn[dbname].assets.delete_many({})
     connection.conn[dbname].metadata.delete_many({})
+    connection.conn[dbname].utxos.delete_many({})
 
 
 @flush_db.register(LocalMongoDBConnection)
@@ -55,6 +56,7 @@ def flush_localmongo_db(connection, dbname):
     connection.conn[dbname].transactions.delete_many({})
     connection.conn[dbname].assets.delete_many({})
     connection.conn[dbname].metadata.delete_many({})
+    connection.conn[dbname].utxos.delete_many({})
 
 
 @singledispatch