Make missing txs check atomic

internal/consensus/replay_stubs.go

@@ -24,6 +24,10 @@ func (m emptyMempool) GetTxsForKeys(txKeys []types.TxKey) types.Txs {
 	return types.Txs{}
 }
 
+func (m emptyMempool) SafeGetTxsForKeys(txKeys []types.TxKey) (types.Txs, []types.TxKey) {
+	return types.Txs{}, []types.TxKey{}
+}
+
 var _ mempool.Mempool = emptyMempool{}
 
 func (emptyMempool) TxStore() *mempool.TxStore { return nil }

internal/consensus/state.go

@@ -2502,13 +2502,12 @@ func (cs *State) tryCreateProposalBlock(ctx context.Context, height int64, round
 // Build a proposal block from mempool txs. If cs.config.GossipTransactionKeyOnly=true
 // proposals only contain txKeys so we rebuild the block using mempool txs
 func (cs *State) buildProposalBlock(height int64, header types.Header, lastCommit *types.Commit, evidence []types.Evidence, proposerAddress types.Address, txKeys []types.TxKey) *types.Block {
-	missingTxs := cs.blockExec.GetMissingTxs(txKeys)
+	txs, missingTxs := cs.blockExec.SafeGetTxsByKeys(txKeys)
 	if len(missingTxs) > 0 {
 		cs.metrics.ProposalMissingTxs.Set(float64(len(missingTxs)))
 		cs.logger.Debug("Missing txs when trying to build block", "missing_txs", cs.blockExec.GetMissingTxs(txKeys))
 		return nil
 	}
-	txs := cs.blockExec.GetTxsForKeys(txKeys)
 	block := cs.state.MakeBlock(height, cs.blockExec.GetTxsForKeys(txKeys), lastCommit, evidence, proposerAddress)
 	block.Version = header.Version
 	block.Data.Txs = txs

internal/mempool/mempool.go

@@ -405,6 +405,23 @@ func (txmp *TxMempool) GetTxsForKeys(txKeys []types.TxKey) types.Txs {
 	return txs
 }
 
+func (txmp *TxMempool) SafeGetTxsForKeys(txKeys []types.TxKey) (types.Txs, []types.TxKey) {
+	txmp.mtx.RLock()
+	defer txmp.mtx.RUnlock()
+
+	txs := make([]types.Tx, 0, len(txKeys))
+	missing := []types.TxKey{}
+	for _, txKey := range txKeys {
+		wtx := txmp.txStore.GetTxByHash(txKey)
+		if wtx == nil {
+			missing = append(missing, txKey)
+			continue
+		}
+		txs = append(txs, wtx.tx)
+	}
+	return txs, missing
+}
+
 // Flush empties the mempool. It acquires a read-lock, fetches all the
 // transactions currently in the transaction store and removes each transaction
 // from the store and all indexes and finally resets the cache.

internal/mempool/types.go

@@ -42,6 +42,10 @@ type Mempool interface {
 
 	GetTxsForKeys(txKeys []types.TxKey) types.Txs
 
+	// Similar to GetTxsForKeys except that it would return a list
+	// indicating missing keys.
+	SafeGetTxsForKeys(txKeys []types.TxKey) (types.Txs, []types.TxKey)
+
 	// ReapMaxBytesMaxGas reaps transactions from the mempool up to maxBytes
 	// bytes total with the condition that the total gasWanted must be less than
 	// maxGas and that the total estimated gas used is less than maxGasEstimated.

internal/state/execution.go

@@ -477,6 +477,10 @@ func (blockExec *BlockExecutor) GetMissingTxs(txKeys []types.TxKey) []types.TxKe
 	return missingTxKeys
 }
 
+func (blockExec *BlockExecutor) SafeGetTxsByKeys(txKeys []types.TxKey) (types.Txs, []types.TxKey) {
+	return blockExec.mempool.SafeGetTxsForKeys(txKeys)
+}
+
 func (blockExec *BlockExecutor) CheckTxFromPeerProposal(ctx context.Context, tx types.Tx) {
 	// Ignore errors from CheckTx because there could be benign errors due to the same tx being
 	// inserted into the mempool from gossiping. Since such simultaneous insertion could result in

internal/state/execution_test.go

@@ -531,7 +531,7 @@ func TestFinalizeBlockValidatorUpdates(t *testing.T) {
 		mock.Anything,
 		mock.Anything,
 		mock.Anything).Return(nil)
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs{})
 	mp.On("TxStore").Return(nil)
 
 	eventBus := eventbus.NewDefault(logger)
@@ -669,7 +669,7 @@ func TestEmptyPrepareProposal(t *testing.T) {
 		mock.Anything,
 		mock.Anything,
 		mock.Anything).Return(nil)
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs{})
 	mp.On("TxStore").Return(nil)
 
 	blockExec := sm.NewBlockExecutor(
@@ -708,7 +708,7 @@ func TestPrepareProposalErrorOnNonExistingRemoved(t *testing.T) {
 	evpool.On("PendingEvidence", mock.Anything).Return([]types.Evidence{}, int64(0))
 
 	mp := &mpmocks.Mempool{}
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs{})
 
 	app := abcimocks.NewApplication(t)
 
@@ -766,7 +766,7 @@ func TestPrepareProposalReorderTxs(t *testing.T) {
 
 	txs := factory.MakeNTxs(height, 10)
 	mp := &mpmocks.Mempool{}
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs(txs))
 
 	trs := txsToTxRecords(types.Txs(txs))
 	trs = trs[2:]
@@ -828,7 +828,7 @@ func TestPrepareProposalErrorOnTooManyTxs(t *testing.T) {
 	maxDataBytes := types.MaxDataBytes(state.ConsensusParams.Block.MaxBytes, 0, nValidators)
 	txs := factory.MakeNTxs(height, maxDataBytes/bytesPerTx+2) // +2 so that tx don't fit
 	mp := &mpmocks.Mempool{}
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs(txs))
 
 	trs := txsToTxRecords(types.Txs(txs))
 
@@ -880,7 +880,7 @@ func TestPrepareProposalErrorOnPrepareProposalError(t *testing.T) {
 
 	txs := factory.MakeNTxs(height, 10)
 	mp := &mpmocks.Mempool{}
-	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs(txs))
+	mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs(txs))
 
 	cm := &abciclientmocks.Client{}
 	cm.On("IsRunning").Return(true)
@@ -984,7 +984,7 @@ func TestCreateProposalAbsentVoteExtensions(t *testing.T) {
 				mock.Anything,
 				mock.Anything,
 				mock.Anything).Return(nil)
-			mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything).Return(types.Txs{})
+			mp.On("ReapMaxBytesMaxGas", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(types.Txs{})
 
 			blockExec := sm.NewBlockExecutor(
 				stateStore,