Problem: Changing validators requires a network restart (#2070)

Solution: Allow nodes to add, update, or remove validators at runtime using a new command. Implements BEP3.

bigchaindb/backend/localmongodb/query.py

@@ -4,10 +4,12 @@
 
 from bigchaindb import backend
 from bigchaindb.backend.exceptions import DuplicateKeyError
+from bigchaindb.common.exceptions import MultipleValidatorOperationError
 from bigchaindb.backend.utils import module_dispatch_registrar
 from bigchaindb.backend.localmongodb.connection import LocalMongoDBConnection
 from bigchaindb.common.transaction import Transaction
 from bigchaindb.backend import mongodb
+from bigchaindb.backend.query import VALIDATOR_UPDATE_ID
 
 register_query = module_dispatch_registrar(backend.query)
 
@@ -262,3 +264,28 @@ def get_unspent_outputs(conn, *, query=None):
         query = {}
     return conn.run(conn.collection('utxos').find(query,
                                                   projection={'_id': False}))
+
+
+@register_query(LocalMongoDBConnection)
+def store_validator_update(conn, validator_update):
+    try:
+        return conn.run(
+            conn.collection('validators')
+            .insert_one(validator_update))
+    except DuplicateKeyError:
+        raise MultipleValidatorOperationError('Validator update already exists')
+
+
+@register_query(LocalMongoDBConnection)
+def get_validator_update(conn, update_id=VALIDATOR_UPDATE_ID):
+    return conn.run(
+        conn.collection('validators')
+        .find_one({'update_id': update_id}, projection={'_id': False}))
+
+
+@register_query(LocalMongoDBConnection)
+def delete_validator_update(conn, update_id=VALIDATOR_UPDATE_ID):
+    return conn.run(
+        conn.collection('validators')
+        .delete_one({'update_id': update_id})
+    )

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', 'utxos', 'assets', 'blocks', 'metadata']:
+    for table_name in ['transactions', 'utxos', 'assets', 'blocks', 'metadata', 'validators']:
         logger.info('Create `%s` table.', table_name)
         # create the table
         # TODO: read and write concerns can be declared here
@@ -41,6 +41,7 @@ def create_indexes(conn, dbname):
     create_blocks_secondary_index(conn, dbname)
     create_metadata_secondary_index(conn, dbname)
     create_utxos_secondary_index(conn, dbname)
+    create_validators_secondary_index(conn, dbname)
 
 
 @register_schema(LocalMongoDBConnection)
@@ -110,3 +111,11 @@ def create_utxos_secondary_index(conn, dbname):
         name='utxo',
         unique=True,
     )
+
+
+def create_validators_secondary_index(conn, dbname):
+    logger.info('Create `validators` secondary index.')
+
+    conn.conn[dbname]['validators'].create_index('update_id',
+                                                 name='update_id',
+                                                 unique=True,)

bigchaindb/backend/query.py

@@ -4,6 +4,8 @@
 
 from bigchaindb.backend.exceptions import OperationError
 
+VALIDATOR_UPDATE_ID = 'a_unique_id_string'
+
 
 @singledispatch
 def write_transaction(connection, signed_transaction):
@@ -606,3 +608,24 @@ def get_unspent_outputs(connection, *, query=None):
     """
 
     raise NotImplementedError
+
+
+@singledispatch
+def store_validator_update(conn, validator_update):
+    """Store a update for the validator set """
+
+    raise NotImplementedError
+
+
+@singledispatch
+def get_validator_update(conn):
+    """Get validator updates which are not synced"""
+
+    raise NotImplementedError
+
+
+@singledispatch
+def delete_validator_update(conn, id):
+    """Set the sync status for validator update documents"""
+
+    raise NotImplementedError

bigchaindb/commands/bigchaindb.py

@@ -10,14 +10,16 @@
 import sys
 
 from bigchaindb.common.exceptions import (DatabaseAlreadyExists,
-                                          DatabaseDoesNotExist)
+                                          DatabaseDoesNotExist,
+                                          MultipleValidatorOperationError)
 import bigchaindb
 from bigchaindb import backend
 from bigchaindb.backend import schema
 from bigchaindb.backend import query
 from bigchaindb.commands import utils
 from bigchaindb.commands.utils import (
     configure_bigchaindb, start_logging_process, input_on_stderr)
+from bigchaindb.backend.query import VALIDATOR_UPDATE_ID
 
 
 logging.basicConfig(level=logging.INFO)
@@ -92,6 +94,24 @@ def run_configure(args):
     print('Ready to go!', file=sys.stderr)
 
 
+@configure_bigchaindb
+def run_upsert_validator(args):
+    """Store validators which should be synced with Tendermint"""
+
+    b = bigchaindb.Bigchain()
+    validator = {'pub_key': {'type': 'ed25519',
+                             'data': args.public_key},
+                 'power': args.power}
+    validator_update = {'validator': validator,
+                        'update_id': VALIDATOR_UPDATE_ID}
+    try:
+        query.store_validator_update(b.connection, validator_update)
+    except MultipleValidatorOperationError:
+        logger.error('A validator update is pending to be applied. '
+                     'Please re-try after the current update has '
+                     'been processed.')
+
+
 def _run_init():
     bdb = bigchaindb.Bigchain()
 
@@ -177,6 +197,7 @@ def create_parser():
     # parser for writing a config file
     config_parser = subparsers.add_parser('configure',
                                           help='Prepare the config file.')
+
     config_parser.add_argument('backend',
                                choices=['localmongodb'],
                                default='localmongodb',
@@ -185,6 +206,17 @@ def create_parser():
                                help='The backend to use. It can only be '
                                '"localmongodb", currently.')
 
+    validator_parser = subparsers.add_parser('upsert-validator',
+                                             help='Add/update/delete a validator')
+
+    validator_parser.add_argument('public_key',
+                                  help='Public key of the validator.')
+
+    validator_parser.add_argument('power',
+                                  type=int,
+                                  help='Voting power of the validator. '
+                                  'Setting it to 0 will delete the validator.')
+
     # parsers for showing/exporting config values
     subparsers.add_parser('show-config',
                           help='Show the current configuration')

bigchaindb/common/exceptions.py

@@ -114,3 +114,7 @@ class ThresholdTooDeep(ValidationError):
 
 class GenesisBlockAlreadyExistsError(ValidationError):
     """Raised when trying to create the already existing genesis block"""
+
+
+class MultipleValidatorOperationError(ValidationError):
+    """Raised when a validator update pending but new request is submited"""

bigchaindb/tendermint/core.py

@@ -3,7 +3,7 @@
 import logging
 
 from abci import BaseApplication, Result
-from abci.types_pb2 import ResponseEndBlock, ResponseInfo
+from abci.types_pb2 import ResponseEndBlock, ResponseInfo, Validator
 
 from bigchaindb.tendermint import BigchainDB
 from bigchaindb.tendermint.utils import decode_transaction, calculate_hash
@@ -106,7 +106,17 @@ def end_block(self, height):
         else:
             self.block_txn_hash = block['app_hash']
 
-        return ResponseEndBlock()
+        validator_updates = self.bigchaindb.get_validator_update()
+        validator_updates = [encode_validator(v) for v in validator_updates]
+
+        # set sync status to true
+        self.bigchaindb.delete_validator_update()
+
+        # NOTE: interface for `ResponseEndBlock` has be changed in the latest
+        # version of py-abci i.e. the validator updates should be return
+        # as follows:
+        # ResponseEndBlock(validator_updates=validator_updates)
+        return ResponseEndBlock(diffs=validator_updates)
 
     def commit(self):
         """Store the new height and along with block hash."""
@@ -121,3 +131,12 @@ def commit(self):
 
         data = self.block_txn_hash.encode('utf-8')
         return Result.ok(data=data)
+
+
+def encode_validator(v):
+    pub_key = v['pub_key']['data']
+    # NOTE: tendermint expects public to be encoded in go-wire format
+    # so `01` has to be appended
+    pubKey = bytes.fromhex('01{}'.format(pub_key))
+    return Validator(pubKey=pubKey,
+                     power=v['power'])

bigchaindb/tendermint/lib.py

@@ -24,7 +24,6 @@
 from bigchaindb.tendermint import fastquery
 from bigchaindb import exceptions as core_exceptions
 
-
 logger = logging.getLogger(__name__)
 
 BIGCHAINDB_TENDERMINT_HOST = getenv('BIGCHAINDB_TENDERMINT_HOST',
@@ -346,5 +345,12 @@ def get_validators(self):
             logger.error('Error while connecting to Tendermint HTTP API')
             raise e
 
+    def get_validator_update(self):
+        update = backend.query.get_validator_update(self.connection)
+        return [update['validator']] if update else []
+
+    def delete_validator_update(self):
+        return backend.query.delete_validator_update(self.connection)
+
 
 Block = namedtuple('Block', ('app_hash', 'height', 'transactions'))

tests/backend/localmongodb/test_queries.py

@@ -310,3 +310,24 @@ def test_get_unspent_outputs(db_context, utxoset):
     assert retrieved_utxoset == list(
         utxo_collection.find(projection={'_id': False}))
     assert retrieved_utxoset == unspent_outputs
+
+
+def test_store_validator_update():
+    from bigchaindb.backend import connect, query
+    from bigchaindb.backend.query import VALIDATOR_UPDATE_ID
+    from bigchaindb.common.exceptions import MultipleValidatorOperationError
+
+    conn = connect()
+
+    validator_update = {'validator': {'key': 'value'},
+                        'update_id': VALIDATOR_UPDATE_ID}
+    query.store_validator_update(conn, deepcopy(validator_update))
+
+    with pytest.raises(MultipleValidatorOperationError):
+        query.store_validator_update(conn, deepcopy(validator_update))
+
+    resp = query.get_validator_update(conn, VALIDATOR_UPDATE_ID)
+
+    assert resp == validator_update
+    assert query.delete_validator_update(conn, VALIDATOR_UPDATE_ID)
+    assert not query.get_validator_update(conn, VALIDATOR_UPDATE_ID)

tests/backend/localmongodb/test_schema.py

@@ -19,7 +19,7 @@ def test_init_creates_db_tables_and_indexes():
 
     collection_names = conn.conn[dbname].collection_names()
     assert set(collection_names) == {
-            'transactions', 'assets', 'metadata', 'blocks', 'utxos'}
+            'transactions', 'assets', 'metadata', 'blocks', 'utxos', 'validators'}
 
     indexes = conn.conn[dbname]['assets'].index_information().keys()
     assert set(indexes) == {'_id_', 'asset_id', 'text'}
@@ -34,6 +34,9 @@ def test_init_creates_db_tables_and_indexes():
     indexes = conn.conn[dbname]['utxos'].index_information().keys()
     assert set(indexes) == {'_id_', 'utxo'}
 
+    indexes = conn.conn[dbname]['validators'].index_information().keys()
+    assert set(indexes) == {'_id_', 'update_id'}
+
 
 def test_init_database_fails_if_db_exists():
     import bigchaindb
@@ -66,7 +69,7 @@ def test_create_tables():
 
     collection_names = conn.conn[dbname].collection_names()
     assert set(collection_names) == {
-            'transactions', 'assets', 'metadata', 'blocks', 'utxos'}
+            'transactions', 'assets', 'metadata', 'blocks', 'utxos', 'validators'}
 
 
 def test_create_secondary_indexes():

tests/commands/test_commands.py

@@ -17,6 +17,7 @@ def test_make_sure_we_dont_remove_any_command():
     assert parser.parse_args(['init']).command
     assert parser.parse_args(['drop']).command
     assert parser.parse_args(['start']).command
+    assert parser.parse_args(['upsert-validator', 'TEMP_PUB_KEYPAIR', '10']).command
 
 
 @pytest.mark.tendermint
@@ -363,3 +364,15 @@ def __init__(self, height):
 
     def json(self):
         return {'result': {'latest_block_height': self.height}}
+
+
+@patch('bigchaindb.config_utils.autoconfigure')
+@patch('bigchaindb.backend.query.store_validator_update')
+@pytest.mark.tendermint
+def test_upsert_validator(mock_autoconfigure, mock_store_validator_update):
+    from bigchaindb.commands.bigchaindb import run_upsert_validator
+
+    args = Namespace(public_key='BOB_PUBLIC_KEY', power='10', config={})
+    run_upsert_validator(args)
+
+    assert mock_store_validator_update.called

tests/tendermint/conftest.py

@@ -5,3 +5,8 @@
 def b():
     from bigchaindb.tendermint import BigchainDB
     return BigchainDB()
+
+
+@pytest.fixture
+def validator_pub_key():
+    return 'B0E42D2589A455EAD339A035D6CE1C8C3E25863F268120AA0162AD7D003A4014'

tests/tendermint/test_core.py

@@ -142,3 +142,27 @@ def test_deliver_transfer_tx__double_spend_fails(b):
 
     result = app.deliver_tx(encode_tx_to_bytes(double_spend))
     assert result.is_error()
+
+
+def test_end_block_return_validator_updates(b):
+    from bigchaindb.tendermint import App
+    from bigchaindb.backend import query
+    from bigchaindb.tendermint.core import encode_validator
+    from bigchaindb.backend.query import VALIDATOR_UPDATE_ID
+
+    app = App(b)
+    app.init_chain(['ignore'])
+    app.begin_block('ignore')
+
+    validator = {'pub_key': {'type': 'ed25519',
+                             'data': 'B0E42D2589A455EAD339A035D6CE1C8C3E25863F268120AA0162AD7D003A4014'},
+                 'power': 10}
+    validator_update = {'validator': validator,
+                        'update_id': VALIDATOR_UPDATE_ID}
+    query.store_validator_update(b.connection, validator_update)
+
+    resp = app.end_block(99)
+    assert resp.diffs[0] == encode_validator(validator)
+
+    updates = b.get_validator_update()
+    assert updates == []

tests/tendermint/test_lib.py

@@ -132,6 +132,26 @@ def test_post_transaction_invalid_mode(b):
         b.write_transaction(tx, 'nope')
 
 
+@pytest.mark.bdb
+def test_validator_updates(b, validator_pub_key):
+    from bigchaindb.backend import query
+    from bigchaindb.backend.query import VALIDATOR_UPDATE_ID
+
+    # create a validator update object
+    validator = {'pub_key': {'type': 'ed25519',
+                             'data': validator_pub_key},
+                 'power': 10}
+    validator_update = {'validator': validator,
+                        'update_id': VALIDATOR_UPDATE_ID}
+    query.store_validator_update(b.connection, validator_update)
+
+    updates = b.get_validator_update()
+    assert updates == [validator_update['validator']]
+
+    b.delete_validator_update()
+    assert b.get_validator_update() == []
+
+
 @pytest.mark.bdb
 def test_update_utxoset(tb, signed_create_tx, signed_transfer_tx, db_context):
     mongo_client = MongoClient(host=db_context.host, port=db_context.port)

tests/utils.py

@@ -57,6 +57,7 @@ def flush_localmongo_db(connection, dbname):
     connection.conn[dbname].assets.delete_many({})
     connection.conn[dbname].metadata.delete_many({})
     connection.conn[dbname].utxos.delete_many({})
+    connection.conn[dbname].validators.delete_many({})
 
 
 @singledispatch