#186 Prevent invalid block time from Tendermint

consensus/abci.go

@@ -16,6 +16,7 @@ import (
 	ethTypes "github.com/ethereum/go-ethereum/core/types"
 	tmtAbciTypes "github.com/tendermint/tendermint/abci/types"
 	tmtLog "github.com/tendermint/tendermint/libs/log"
+	"time"
 )
 
 // maxTransactionSize is 32KB in order to prevent DOS attacks
@@ -97,11 +98,24 @@ func (abci *TendermintABCI) InitChain(req tmtAbciTypes.RequestInitChain) tmtAbci
 // for the validators.
 //
 // Response:
-//		- Optional Key-Value tags for filtering and indexing
+// 		- Optional Key-Value tags for filtering and indexing
 func (abci *TendermintABCI) BeginBlock(req tmtAbciTypes.RequestBeginBlock) tmtAbciTypes.ResponseBeginBlock {
-	abci.logger.Debug("Beginning new block", "hash", req.Hash)
-	abci.db.UpdateBlockState(&req.Header)
-
+	abci.logger.Debug("Beginning new block", "hash", req.Hash, "height", req.Header.Height)
+	parentBlock := abci.getCurrentBlock()
+
+	// IMPORTANT: According to Tendermint documentation and based on the consensus setup made for our network
+	// two consecutive blocks cannot generated less than one second apart. BUT we identify an issue on that assumption
+	// and reported here https://github.com/tendermint/tendermint/issues/3755. Therefore we implemented this mitigation code
+	// to prevent ethereum headers to be invalid. Learn more in https://github.com/lightstreams-network/lightchain/issues/186
+	if uint64(req.Header.Time.Unix()) <= parentBlock.Time() {
+		abci.metrics.ReplacedBlockTimeTotal.Add(1)
+		nextBlockTime := time.Unix(int64(parentBlock.Time() + 1), 0)
+		abci.logger.Error(fmt.Sprintf("Invalid consensus BlockTime. Replacing block time %d...", req.Header.Height), 
+			"original", req.Header.Time.Unix(), "replaced", nextBlockTime.Unix())
+		req.Header.Time = nextBlockTime
+	}
+
+	abci.db.UpdateBlockState(req.Header)
 	return tmtAbciTypes.ResponseBeginBlock{}
 }
 
@@ -132,7 +146,7 @@ func (abci *TendermintABCI) CheckTx(txBytes []byte) tmtAbciTypes.ResponseCheckTx
 		return tmtAbciTypes.ResponseCheckTx{Code: 1, Log: "INVALID_TX"}
 	}
 
-	abci.logger.Info("Checking TX", "hash", tx.Hash().String(), "nonce", tx.Nonce(), "cost", tx.Cost())
+	abci.logger.Info("Checking TX", "hash", tx.Hash().String(), "nonce", tx.Nonce(), "cost", tx.Cost(), "height", abci.db.GetBlockStateHeader().Number.Uint64())
 
 	var signer ethTypes.Signer = ethTypes.FrontierSigner{}
 	if tx.Protected() {
@@ -226,7 +240,8 @@ func (abci *TendermintABCI) DeliverTx(txBytes []byte) tmtAbciTypes.ResponseDeliv
 		return tmtAbciTypes.ResponseDeliverTx{Code: 1, Log: "INVALID_TX"}
 	}
 
-	abci.logger.Info("Delivering TX", "hash", tx.Hash().String(), "nonce", tx.Nonce(), "cost", tx.Cost(), "gas", tx.Gas(), "gas_price", tx.GasPrice())
+	abci.logger.Info("Delivering TX", "hash", tx.Hash().String(), "nonce", tx.Nonce(), "cost", tx.Cost(), 
+		"gas", tx.Gas(), "height", abci.db.GetBlockStateHeader().Number.Uint64())
 
 	res := abci.db.ExecuteTx(tx)
 	if res.IsErr() {
@@ -270,7 +285,7 @@ func (abci *TendermintABCI) Commit() tmtAbciTypes.ResponseCommit {
 		abci.metrics.CommitErrBlockTotal.Add(1, "UNABLE_TO_PERSIST")
 		panic(err)
 	}
-
+  
 	ethState, err := abci.getCurrentDBState()
 	if err != nil {
 		abci.logger.Error("Error getting next latest state", "err", err)

consensus/metrics/init.go

@@ -17,32 +17,35 @@ const errorCodeLabel = "error_code"
 
 // Metrics contains metrics exposed by this package.
 type Metrics struct {
-	CheckTxsTotal       CheckTxTotalMetric
-	CheckErrTxsTotal    CheckErrTxsTotalMetric
-	DeliverTxsTotal     DeliverTxsTotalMetric
-	DeliverErrTxsTotal  DeliverErrTxsTotalMetric
-	CommitBlockTotal    CommitBlockTotalMetric
-	CommitErrBlockTotal CommitErrBlockTotalMetric
+	CheckTxsTotal          CheckTxTotalMetric
+	CheckErrTxsTotal       CheckErrTxsTotalMetric
+	DeliverTxsTotal        DeliverTxsTotalMetric
+	DeliverErrTxsTotal     DeliverErrTxsTotalMetric
+	CommitBlockTotal       CommitBlockTotalMetric
+	CommitErrBlockTotal    CommitErrBlockTotalMetric
+	ReplacedBlockTimeTotal ReplacedBlockTimeTotalMetric
 }
 
 func NewMetrics(registry *prometheus.Registry) Metrics {
 	return Metrics{
-		CheckTxsTotal:       NewCheckTxTotalMetric(registry),
-		CheckErrTxsTotal:    NewCheckErrTxsTotalMetric(registry),
-		DeliverTxsTotal:     NewDeliverTxsTotalMetric(registry),
-		DeliverErrTxsTotal:  NewDeliverErrTxsTotalMetric(registry),
-		CommitBlockTotal:    NewCommitBlockTotalMetric(registry),
-		CommitErrBlockTotal: NewCommitErrBlockTotalMetric(registry),
+		CheckTxsTotal:          NewCheckTxTotalMetric(registry),
+		CheckErrTxsTotal:       NewCheckErrTxsTotalMetric(registry),
+		DeliverTxsTotal:        NewDeliverTxsTotalMetric(registry),
+		DeliverErrTxsTotal:     NewDeliverErrTxsTotalMetric(registry),
+		CommitBlockTotal:       NewCommitBlockTotalMetric(registry),
+		CommitErrBlockTotal:    NewCommitErrBlockTotalMetric(registry),
+		ReplacedBlockTimeTotal: NewReplacedBlockTimeTotalMetric(registry),
 	}
 }
 
 func NewNullMetrics() Metrics {
 	return Metrics{
-		CheckTxsTotal:       NewNullCheckTxTotalMetric(),
-		CheckErrTxsTotal:    NewNullCheckErrTrxTotalMetric(),
-		DeliverTxsTotal:     NewNullDeliverTxsTotalMetric(),
-		DeliverErrTxsTotal:  NewNullDeliverErrTxsTotalMetric(),
-		CommitBlockTotal:    NewNullCommitBlockTotalMetric(),
-		CommitErrBlockTotal: NewNullCommitErrBlockTotalMetric(),
+		CheckTxsTotal:          NewNullCheckTxTotalMetric(),
+		CheckErrTxsTotal:       NewNullCheckErrTrxTotalMetric(),
+		DeliverTxsTotal:        NewNullDeliverTxsTotalMetric(),
+		DeliverErrTxsTotal:     NewNullDeliverErrTxsTotalMetric(),
+		CommitBlockTotal:       NewNullCommitBlockTotalMetric(),
+		CommitErrBlockTotal:    NewNullCommitErrBlockTotalMetric(),
+		ReplacedBlockTimeTotal: NewNullReplacedBlockTimeTotalMetric(),
 	}
 }

consensus/metrics/replaced_block_time_total_metric.go

@@ -0,0 +1,34 @@
+package metrics
+
+import (
+	kitprometheus "github.com/go-kit/kit/metrics/prometheus"
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/go-kit/kit/metrics"
+	"github.com/go-kit/kit/metrics/discard"
+)
+
+type ReplacedBlockTimeTotalMetric struct {
+	metrics.Counter
+}
+
+func NewReplacedBlockTimeTotalMetric(registry *prometheus.Registry) ReplacedBlockTimeTotalMetric {
+	metric := prometheus.NewCounterVec(prometheus.CounterOpts{
+		Namespace:   namespace,
+		Subsystem:   metricsSubsystem,
+		Name:        "replaced_block_time_total_counter",
+		Help:        "Replaced consensus block time total.",
+		ConstLabels: moduleAbciConstLabelValues,
+	}, []string{})
+
+	registry.Register(metric)
+
+	return ReplacedBlockTimeTotalMetric{
+		kitprometheus.NewCounter(metric).With(),
+	}
+}
+
+func NewNullReplacedBlockTimeTotalMetric() ReplacedBlockTimeTotalMetric {
+	return ReplacedBlockTimeTotalMetric{
+		discard.NewCounter(),
+	}
+}

database/blockstate.go

@@ -88,10 +88,10 @@ func (bs *blockState) persist(bc *core.BlockChain, db ethdb.Database) (ethTypes.
 	return *block, nil
 }
 
-func (bs *blockState) updateBlockState(config *params.ChainConfig, parentTime uint64, numTx uint64) {
+func (bs *blockState) updateBlockState(config params.ChainConfig, blockTime uint64, numTx uint64) {
 	parentHeader := bs.parent.Header()
-	bs.header.Time = new(big.Int).SetUint64(parentTime).Uint64()
-	bs.header.Difficulty = ethash.CalcDifficulty(config, parentTime, parentHeader)
+	bs.header.Time = new(big.Int).SetUint64(blockTime).Uint64()
+	bs.header.Difficulty = ethash.CalcDifficulty(&config, blockTime, parentHeader)
 	bs.transactions = make([]*ethTypes.Transaction, 0, numTx)
 	bs.receipts = make([]*ethTypes.Receipt, 0, numTx)
 }

database/database.go

@@ -98,15 +98,19 @@ func (db *Database) ResetBlockState(receiver common.Address) error {
 }
 
 // UpdateBlockState uses the tendermint header to update the eth header.
-func (db *Database) UpdateBlockState(tmHeader *tmtAbciTypes.Header) {
+func (db *Database) UpdateBlockState(tmHeader tmtAbciTypes.Header) {
 	db.logger.Debug("Updating DB BlockState")
 	db.ethState.UpdateBlockState(
-		db.eth.APIBackend.ChainConfig(),
+		*db.eth.APIBackend.ChainConfig(),
 		uint64(tmHeader.Time.Unix()),
 		uint64(tmHeader.GetNumTxs()),
 	)
 }
 
+func (db *Database) GetBlockStateHeader() ethTypes.Header {
+	return *db.ethState.blockState.header
+}
+
 // GasLimit returns the maximum gas per block.
 func (db *Database) GasLimit() uint64 {
 	return db.ethState.GasLimit().Gas()

database/state.go

@@ -14,7 +14,7 @@ import (
 	tmtLog "github.com/tendermint/tendermint/libs/log"
 )
 
-//----------------------------------------------------------------------
+// ----------------------------------------------------------------------
 // EthState manages concurrent access to the intermediate blockState object
 // The eth tx pool fires TxPreEvent in a go-routine,
 // and the miner subscribes to this in another go-routine and processes the tx onto
@@ -40,7 +40,7 @@ func NewEthState(ethereum *eth.Ethereum, ethCfg *eth.Config, logger tmtLog.Logge
 	return &EthState{
 		ethereum:  ethereum,
 		ethConfig: ethCfg,
-		logger: logger,
+		logger:    logger,
 	}
 }
 
@@ -108,11 +108,11 @@ func (es *EthState) resetBlockState(receiver common.Address) error {
 	return nil
 }
 
-func (es *EthState) UpdateBlockState(config *params.ChainConfig, parentTime uint64, numTx uint64) {
+func (es *EthState) UpdateBlockState(config params.ChainConfig, blockTime uint64, numTx uint64) {
 	es.mtx.Lock()
 	defer es.mtx.Unlock()
 
-	es.blockState.updateBlockState(config, parentTime, numTx)
+	es.blockState.updateBlockState(config, blockTime, numTx)
 }
 
 func (es *EthState) GasLimit() *core.GasPool {
@@ -141,4 +141,4 @@ func newBlockHeader(receiver common.Address, prevBlock *ethTypes.Block) *ethType
 		GasLimit:   core.CalcGasLimit(prevBlock, prevBlock.GasLimit(), prevBlock.GasLimit()),
 		Coinbase:   receiver,
 	}
-}
+}

network/network.go

@@ -130,7 +130,10 @@ func createConsensusGenesis(pv *privval.FilePV) ([]byte, error) {
 		PubKey:  pv.GetPubKey(),
 		Power:   10,
 	}}
-
+
+	// Using less than 1 second TimeIota to simulate a consensus invalid block time 
+	genDoc.ConsensusParams.Block.TimeIotaMs = 950
+
 	genDocBytes, err := cdc.MarshalJSONIndent(genDoc, "", "  ")
 	if err != nil {
 		return nil, err

network/standalone/consensus/config.go

@@ -137,7 +137,7 @@ max_num_inbound_peers = 40
 max_num_outbound_peers = 10
 
 # Time to wait before flushing messages out on the connection
-flush_throttle_timeout = "100ms"
+flush_throttle_timeout = "35ms"
 
 # Maximum size of a message packet payload, in bytes
 max_packet_msg_payload_size = 1024
@@ -164,8 +164,8 @@ private_peer_ids = ""
 allow_duplicate_ip = true
 
 # Peer connection configuration.
-handshake_timeout = "20s"
-dial_timeout = "3s"
+handshake_timeout = "5s"
+dial_timeout = "1s"
 
 ##### mempool configuration options #####
 [mempool]
@@ -185,27 +185,27 @@ cache_size = 10000
 
 wal_file = "data/cs.wal/wal"
 
-timeout_propose = "8s"
-timeout_propose_delta = "500ms"
-timeout_prevote = "2s"
-timeout_prevote_delta = "500ms"
-timeout_precommit = "2s"
-timeout_precommit_delta = "500ms"
-timeout_commit = "2s"
+timeout_propose = "500ms"
+timeout_propose_delta = "100ms"
+timeout_prevote = "500ms"
+timeout_prevote_delta = "100ms"
+timeout_precommit = "500ms"
+timeout_precommit_delta = "100ms"
+timeout_commit = "500ms"
 
 # Make progress as soon as we have all the precommits (as if TimeoutCommit = 0)
 skip_timeout_commit = false
 
 # EmptyBlocks mode and possible interval between empty blocks
 create_empty_blocks = true
-create_empty_blocks_interval = "10s"
+create_empty_blocks_interval = "5s"
 
 # Reactor sleep duration parameters
 peer_gossip_sleep_duration = "100ms"
-peer_query_maj23_sleep_duration = "2s"
+peer_query_maj23_sleep_duration = "1s"
 
 # Block time parameters. Corresponds to the minimum time increment between consecutive blocks.
-blocktime_iota = "1s"
+blocktime_iota = "950ms"
 
 ##### transactions indexer configuration options #####
 [tx_index]

scripts/init.sh

@@ -76,7 +76,7 @@ fi
 run "$EXEC_CMD"
 
 echo "Restoring ${NETWORK} private keys"
-run "cp ./network/${NETWORK}/database/keystore/* ${DATA_DIR}/database/keystore/"		
+run "rsync -avzh --ignore-errors  ./network/${NETWORK}/database/keystore/ ${DATA_DIR}/database/keystore"
 
 popd
 

scripts/run.sh

@@ -66,7 +66,7 @@ if [ -n "${CLEAN}" ]; then
 	    run "rm -rf ${DATA_DIR}"
 		run "$EXEC_BIN init ${INIT_ARGS}"
 		echo "Restoring ${NETWORK} private keys"
-		run "cp ./network/${NETWORK}/database/keystore/* ${DATA_DIR}/database/keystore/"		
+		run "rsync -avzh --ignore-errors ./network/${NETWORK}/database/keystore/ ${DATA_DIR}/database/keystore"
 		echo -e "################################ \n"
 	else
 		echo -e "Exiting"