[FAB-1038] Rework committer to be more general

While interating over block transaction and running
validation, need also to check the uniqueness of txid.
This commit adds check to query ledger for transactions
to verify that there is no duplicates.

Change-Id: Ieea4c9e61d0a320d7842ba3ab21f6d50e583beba
Signed-off-by: Artem Barger <bartem@il.ibm.com>

core/committer/txvalidator/txvalidator_test.go

@@ -19,11 +19,14 @@ package txvalidator
 import (
 	"testing"
 
+	"github.com/golang/protobuf/proto"
 	"github.com/hyperledger/fabric/core/ledger/ledgermgmt"
 	"github.com/hyperledger/fabric/core/ledger/testutil"
 	"github.com/hyperledger/fabric/core/ledger/util"
+	util2 "github.com/hyperledger/fabric/core/util"
 	"github.com/hyperledger/fabric/protos/common"
 	pb "github.com/hyperledger/fabric/protos/peer"
+	"github.com/hyperledger/fabric/protos/utils"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
 )
@@ -65,3 +68,65 @@ func TestKVLedgerBlockStorage(t *testing.T) {
 	assert.True(t, !txsfltr.IsSet(uint(1)))
 	assert.True(t, !txsfltr.IsSet(uint(2)))
 }
+
+func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
+	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
+	ledgermgmt.InitializeTestEnv()
+	defer ledgermgmt.CleanupTestEnv()
+	ledger, _ := ledgermgmt.CreateLedger("TestLedger")
+	defer ledger.Close()
+
+	validator := &txValidator{ledger, &mockVsccValidator{}}
+
+	// Create simeple endorsement transaction
+	payload := &common.Payload{
+		Header: &common.Header{
+			ChainHeader: &common.ChainHeader{
+				TxID:    "simple_txID", // Fake txID
+				Type:    int32(common.HeaderType_ENDORSER_TRANSACTION),
+				ChainID: util2.GetTestChainID(),
+			},
+		},
+		Data: []byte("test"),
+	}
+
+	payloadBytes, err := proto.Marshal(payload)
+
+	// Check marshaling didn't fail
+	assert.NoError(t, err)
+
+	// Envelope the payload
+	envelope := &common.Envelope{
+		Payload: payloadBytes,
+	}
+
+	envelopeBytes, err := proto.Marshal(envelope)
+
+	// Check marshaling didn't fail
+	assert.NoError(t, err)
+
+	block := &common.Block{
+		Data: &common.BlockData{
+			// Enconde transactions
+			Data: [][]byte{envelopeBytes},
+		},
+	}
+
+	block.Header = &common.BlockHeader{
+		Number:   1,
+		DataHash: block.Data.Hash(),
+	}
+
+	// Initialize metadata
+	utils.InitBlockMetadata(block)
+	// Commit block to the ledger
+	ledger.Commit(block)
+
+	// Validation should invalidate transaction,
+	// because it's already committed
+	validator.Validate(block)
+
+	txsfltr := util.NewFilterBitArrayFromBytes(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+
+	assert.True(t, txsfltr.IsSet(0))
+}

core/committer/txvalidator/validator.go

@@ -77,10 +77,12 @@ func (v *txValidator) Validate(block *common.Block) {
 	defer logger.Debug("END Block Validation")
 	txsfltr := ledgerUtil.NewFilterBitArray(uint(len(block.Data.Data)))
 	for tIdx, d := range block.Data.Data {
+		// Start by marking transaction as invalid, before
+		// doing any validation checks.
+		txsfltr.Set(uint(tIdx))
 		if d != nil {
 			if env, err := utils.GetEnvelopeFromBlock(d); err != nil {
 				logger.Warningf("Error getting tx from block(%s)", err)
-				txsfltr.Set(uint(tIdx))
 			} else if env != nil {
 				// validate the transaction: here we check that the transaction
 				// is properly formed, properly signed and that the security
@@ -90,25 +92,32 @@ func (v *txValidator) Validate(block *common.Block) {
 				logger.Debug("Validating transaction peer.ValidateTransaction()")
 				if payload, _, err := peer.ValidateTransaction(env); err != nil {
 					logger.Errorf("Invalid transaction with index %d, error %s", tIdx, err)
-					txsfltr.Set(uint(tIdx))
 				} else {
+					// Check duplicate transactions
+					txID := payload.Header.ChainHeader.TxID
+					if _, err := v.ledger.GetTransactionByID(txID); err == nil {
+						logger.Warning("Duplicate transaction found, ", txID, ", skipping")
+						continue
+					}
+
 					//the payload is used to get headers
 					logger.Debug("Validating transaction vscc tx validate")
 					if err = v.vscc.VSCCValidateTx(payload, d); err != nil {
-						txID := payload.Header.ChainHeader.TxID
+						txID := txID
 						logger.Errorf("VSCCValidateTx for transaction txId = %s returned error %s", txID, err)
-						txsfltr.Set(uint(tIdx))
 						continue
 					}
 
 					if _, err := proto.Marshal(env); err != nil {
-						logger.Warningf("Cannot marshal transactoins %s", err)
-						txsfltr.Set(uint(tIdx))
+						logger.Warningf("Cannot marshal transaction due to %s", err)
+						continue
 					}
+					// Succeeded to pass down here, transaction is valid,
+					// just unset the filter bit array flag.
+					txsfltr.Unset(uint(tIdx))
 				}
 			} else {
 				logger.Warning("Nil tx from block")
-				txsfltr.Set(uint(tIdx))
 			}
 		}
 	}