[FAB-10201] handle duplicate tx detection errors

GetTransactionByID is used to detect duplicate txids in the ledger. This
function can return an error if there was an I/O problem (and so the duplicate
status of the tx couldn't be properly judged). In that case we must
abort processing on that chain, lest we suffer a state fork.

Change-Id: I32a8ea9c0803f7a38b4c6eb3ea30b4ce1a651456
Signed-off-by: Alessandro Sorniotti <ale.linux@sopit.net>

common/ledger/blkstorage/blockstorage.go

@@ -20,6 +20,7 @@ import (
 	"errors"
 
 	"github.com/hyperledger/fabric/common/ledger"
+	l "github.com/hyperledger/fabric/core/ledger"
 	"github.com/hyperledger/fabric/protos/common"
 	"github.com/hyperledger/fabric/protos/peer"
 )
@@ -44,7 +45,8 @@ type IndexConfig struct {
 
 var (
 	// ErrNotFoundInIndex is used to indicate missing entry in the index
-	ErrNotFoundInIndex = errors.New("Entry not found in index")
+	ErrNotFoundInIndex = l.NotFoundInIndexErr("")
+
 	// ErrAttrNotIndexed is used to indicate that an attribute is not indexed
 	ErrAttrNotIndexed = errors.New("Attribute not indexed")
 )

core/committer/txvalidator/txvalidator_test.go

@@ -200,7 +200,7 @@ func TestVeryLargeParallelBlockValidation(t *testing.T) {
 	testValidationWithNTXes(t, ledger, gbHash, 4096)
 }
 
-func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
+func TestTxValidationFailure_InvalidTxid(t *testing.T) {
 	viper.Set("peer.fileSystemPath", "/tmp/fabric/txvalidatortest")
 	ledgermgmt.InitializeTestEnv()
 	defer ledgermgmt.CleanupTestEnv()
@@ -216,14 +216,23 @@ func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
 	}{&mocktxvalidator.Support{LedgerVal: ledger, ACVal: &config.MockApplicationCapabilities{}}, semaphore.NewWeighted(10)}
 	tValidator := &TxValidator{vcs, &validator.MockVsccValidator{}}
 
+	mockSigner, err := mspmgmt.GetLocalMSP().GetDefaultSigningIdentity()
+	assert.NoError(t, err)
+	mockSignerSerialized, err := mockSigner.Serialize()
+	assert.NoError(t, err)
+
 	// Create simple endorsement transaction
 	payload := &common.Payload{
 		Header: &common.Header{
 			ChannelHeader: utils.MarshalOrPanic(&common.ChannelHeader{
-				TxId:      "simple_txID", // Fake txID
+				TxId:      "INVALID TXID!!!",
 				Type:      int32(common.HeaderType_ENDORSER_TRANSACTION),
 				ChannelId: util2.GetTestChainID(),
 			}),
+			SignatureHeader: utils.MarshalOrPanic(&common.SignatureHeader{
+				Nonce:   []byte("nonce"),
+				Creator: mockSignerSerialized,
+			}),
 		},
 		Data: []byte("test"),
 	}
@@ -233,9 +242,13 @@ func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
 	// Check marshaling didn't fail
 	assert.NoError(t, err)
 
+	sig, err := mockSigner.Sign(payloadBytes)
+	assert.NoError(t, err)
+
 	// Envelope the payload
 	envelope := &common.Envelope{
-		Payload: payloadBytes,
+		Payload:   payloadBytes,
+		Signature: sig,
 	}
 
 	envelopeBytes, err := proto.Marshal(envelope)
@@ -271,6 +284,9 @@ func TestNewTxValidator_DuplicateTransactions(t *testing.T) {
 
 	txsfltr := util.TxValidationFlags(block.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
 	assert.True(t, txsfltr.IsInvalid(0))
+
+	// We expect the tx to be invalid because of a bad txid
+	assert.True(t, txsfltr.Flag(0) == peer.TxValidationCode_BAD_PROPOSAL_TXID)
 }
 
 func createCCUpgradeEnvelope(chainID, chaincodeName, chaincodeVersion string, signer msp.SigningIdentity) (*common.Envelope, error) {

core/committer/txvalidator/validator.go

@@ -329,7 +329,10 @@ func (v *TxValidator) validateTx(req *blockValidationRequest, results chan<- *bl
 		if common.HeaderType(chdr.Type) == common.HeaderType_ENDORSER_TRANSACTION {
 			// Check duplicate transactions
 			txID = chdr.TxId
-			if _, err := v.Support.Ledger().GetTransactionByID(txID); err == nil {
+			// GetTransactionByID will return:
+			_, err := v.Support.Ledger().GetTransactionByID(txID)
+			// 1) err == nil => there is already a tx in the ledger with the supplied id
+			if err == nil {
 				logger.Error("Duplicate transaction found, ", txID, ", skipping")
 				results <- &blockValidationResult{
 					tIdx:           tIdx,
@@ -337,6 +340,16 @@ func (v *TxValidator) validateTx(req *blockValidationRequest, results chan<- *bl
 				}
 				return
 			}
+			// 2) err is not of type blkstorage.NotFoundInIndexErr => we could not verify whether a tx with the supplied id is in the ledger
+			if _, isNotFoundInIndexErrType := err.(ledger.NotFoundInIndexErr); !isNotFoundInIndexErrType {
+				logger.Errorf("Ledger failure while attempting to detect duplicate status for txid %s, err '%s'. Aborting", txID, err)
+				results <- &blockValidationResult{
+					tIdx: tIdx,
+					err:  err,
+				}
+				return
+			}
+			// 3) err is of type blkstorage.NotFoundInIndexErr => there is no tx with the supplied id in the ledger
 
 			// Validate tx with vscc and policy
 			logger.Debug("Validating transaction vscc tx validate")

core/committer/txvalidator/validator_test.go

@@ -769,7 +769,7 @@ func TestLedgerIsNoAvailable(t *testing.T) {
 	ccID := "mycc"
 	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
 
-	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, errors.New("Cannot find the transaction"))
+	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, ledger.NotFoundInIndexErr(""))
 
 	queryExecutor := new(mockQueryExecutor)
 	queryExecutor.On("GetState", mock.Anything, mock.Anything).Return([]byte{}, errors.New("Unable to connect to DB"))
@@ -786,6 +786,59 @@ func TestLedgerIsNoAvailable(t *testing.T) {
 	assertion.NotNil(err.(*commonerrors.VSCCInfoLookupFailureError))
 }
 
+func TestLedgerIsNotAvailableForCheckingTxidDuplicate(t *testing.T) {
+	theLedger := new(mockLedger)
+	vcs := struct {
+		*mocktxvalidator.Support
+		*semaphore.Weighted
+	}{&mocktxvalidator.Support{LedgerVal: theLedger, ACVal: &mockconfig.MockApplicationCapabilities{}}, semaphore.NewWeighted(10)}
+	mp := (&scc.MocksccProviderFactory{}).NewSystemChaincodeProvider()
+	pm := &mocks.PluginMapper{}
+	validator := txvalidator.NewTxValidator(vcs, mp, pm)
+
+	ccID := "mycc"
+	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
+
+	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, errors.New("Unable to connect to DB"))
+
+	b := &common.Block{Data: &common.BlockData{Data: [][]byte{utils.MarshalOrPanic(tx)}}}
+
+	err := validator.Validate(b)
+
+	assertion := assert.New(t)
+	// We expect a validation error because the ledger wasn't ready to tell us whether there was a tx with that ID or not
+	assertion.Error(err)
+}
+
+func TestDuplicateTxId(t *testing.T) {
+	theLedger := new(mockLedger)
+	vcs := struct {
+		*mocktxvalidator.Support
+		*semaphore.Weighted
+	}{&mocktxvalidator.Support{LedgerVal: theLedger, ACVal: &mockconfig.MockApplicationCapabilities{}}, semaphore.NewWeighted(10)}
+	mp := (&scc.MocksccProviderFactory{}).NewSystemChaincodeProvider()
+	pm := &mocks.PluginMapper{}
+	validator := txvalidator.NewTxValidator(vcs, mp, pm)
+
+	ccID := "mycc"
+	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
+
+	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, nil)
+
+	b := &common.Block{Data: &common.BlockData{Data: [][]byte{utils.MarshalOrPanic(tx)}}}
+
+	err := validator.Validate(b)
+
+	assertion := assert.New(t)
+	// We expect no validation error because we simply mark the tx as invalid
+	assertion.NoError(err)
+
+	// We expect the tx to be invalid because of a duplicate txid
+	txsfltr := lutils.TxValidationFlags(b.Metadata.Metadata[common.BlockMetadataIndex_TRANSACTIONS_FILTER])
+	assertion.True(txsfltr.IsInvalid(0))
+	assertion.True(txsfltr.Flag(0) == peer.TxValidationCode_DUPLICATE_TXID)
+}
+
 func TestValidationInvalidEndorsing(t *testing.T) {
 	theLedger := new(mockLedger)
 	vcs := struct {
@@ -805,7 +858,7 @@ func TestValidationInvalidEndorsing(t *testing.T) {
 	ccID := "mycc"
 	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
 
-	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, errors.New("Cannot find the transaction"))
+	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, ledger.NotFoundInIndexErr(""))
 
 	cd := &ccp.ChaincodeData{
 		Name:    ccID,

core/ledger/ledger_interface.go

@@ -316,3 +316,10 @@ type ErrCollectionConfigNotYetAvailable struct {
 func (e *ErrCollectionConfigNotYetAvailable) Error() string {
 	return e.Msg
 }
+
+// NotFoundInIndexErr is used to indicate missing entry in the index
+type NotFoundInIndexErr string
+
+func (NotFoundInIndexErr) Error() string {
+	return "Entry not found in index"
+}