Add function TxIDExists and use in validation path

This commit adds a function in ledger TxIDExists and uses this in
validation path for checking duplicate txids.

The current validation code, in order to checks for a duplicate txid,
uses the API GetTransactionByID on ledger and relies on the error types
returned by this API. For a ledger that is bootstrapped from a snapshot,
this API needs to return an additional error if the txid belongs to a
block that is committed prior to the snapshot height.
Instead of exposing one more error type for validation code, we add an
explicit function for checking the existence of a TxID

Signed-off-by: manish <manish.sethi@gmail.com>

common/ledger/blkstorage/blockfile_mgr.go

@@ -586,6 +586,10 @@ func (mgr *blockfileMgr) retrieveBlocks(startNum uint64) (*blocksItr, error) {
 	return newBlockItr(mgr, startNum), nil
 }
 
+func (mgr *blockfileMgr) txIDExists(txID string) (bool, error) {
+	return mgr.index.txIDExists(txID)
+}
+
 func (mgr *blockfileMgr) retrieveTransactionByID(txID string) (*common.Envelope, error) {
 	logger.Debugf("retrieveTransactionByID() - txId = [%s]", txID)
 	loc, err := mgr.index.getTxLoc(txID)

common/ledger/blkstorage/blockfile_mgr_test.go

@@ -170,6 +170,49 @@ func TestBlockfileMgrBlockchainInfo(t *testing.T) {
 	require.Equal(t, uint64(10), bcInfo.Height)
 }
 
+func TestTxIDExists(t *testing.T) {
+	t.Run("green-path", func(t *testing.T) {
+		env := newTestEnv(t, NewConf(testPath(), 0))
+		defer env.Cleanup()
+
+		blkStore, err := env.provider.Open("testLedger")
+		require.NoError(t, err)
+		defer blkStore.Shutdown()
+
+		blocks := testutil.ConstructTestBlocks(t, 2)
+		for _, blk := range blocks {
+			require.NoError(t, blkStore.AddBlock(blk))
+		}
+
+		for _, blk := range blocks {
+			for i := range blk.Data.Data {
+				txID, err := protoutil.GetOrComputeTxIDFromEnvelope(blk.Data.Data[i])
+				require.NoError(t, err)
+				exists, err := blkStore.TxIDExists(txID)
+				require.NoError(t, err)
+				require.True(t, exists)
+			}
+		}
+		exists, err := blkStore.TxIDExists("non-existant-txid")
+		require.NoError(t, err)
+		require.False(t, exists)
+	})
+
+	t.Run("error-path", func(t *testing.T) {
+		env := newTestEnv(t, NewConf(testPath(), 0))
+		defer env.Cleanup()
+
+		blkStore, err := env.provider.Open("testLedger")
+		require.NoError(t, err)
+		defer blkStore.Shutdown()
+
+		env.provider.Close()
+		exists, err := blkStore.TxIDExists("random")
+		require.EqualError(t, err, "error while trying to check the presence of TXID [random]: internal leveldb error while obtaining db iterator: leveldb: closed")
+		require.False(t, exists)
+	})
+}
+
 func TestBlockfileMgrGetTxById(t *testing.T) {
 	env := newTestEnv(t, NewConf(testPath(), 0))
 	defer env.Cleanup()

common/ledger/blkstorage/blockindex.go

@@ -223,6 +223,24 @@ func (index *blockIndex) getTxValidationCodeByTxID(txID string) (peer.TxValidati
 	return peer.TxValidationCode(v.TxValidationCode), nil
 }
 
+func (index *blockIndex) txIDExists(txID string) (bool, error) {
+	if !index.isAttributeIndexed(IndexableAttrTxID) {
+		return false, ErrAttrNotIndexed
+	}
+	rangeScan := constructTxIDRangeScan(txID)
+	itr, err := index.db.GetIterator(rangeScan.startKey, rangeScan.stopKey)
+	if err != nil {
+		return false, errors.WithMessagef(err, "error while trying to check the presence of TXID [%s]", txID)
+	}
+	defer itr.Release()
+
+	present := itr.Next()
+	if err := itr.Error(); err != nil {
+		return false, errors.Wrapf(err, "error while trying to check the presence of TXID [%s]", txID)
+	}
+	return present, nil
+}
+
 func (index *blockIndex) getTxIDVal(txID string) (*TxIDIndexValue, error) {
 	if !index.isAttributeIndexed(IndexableAttrTxID) {
 		return nil, ErrAttrNotIndexed

common/ledger/blkstorage/blockindex_test.go

@@ -151,6 +151,17 @@ func testBlockIndexSelectiveIndexing(t *testing.T, indexItems []IndexableAttr) {
 			require.Exactly(t, ErrAttrNotIndexed, err)
 		}
 
+		// test txIDExists
+		txid, err = protoutil.GetOrComputeTxIDFromEnvelope(blocks[0].Data.Data[0])
+		require.NoError(t, err)
+		exists, err := blockfileMgr.txIDExists(txid)
+		if containsAttr(indexItems, IndexableAttrTxID) {
+			require.NoError(t, err)
+			require.True(t, exists)
+		} else {
+			require.Exactly(t, ErrAttrNotIndexed, err)
+		}
+
 		//test 'retrieveTrasnactionsByBlockNumTranNum
 		txEnvelope2, err := blockfileMgr.retrieveTransactionByBlockNumTranNum(0, 0)
 		if containsAttr(indexItems, IndexableAttrBlockNumTranNum) {

common/ledger/blkstorage/blockstore.go

@@ -72,6 +72,11 @@ func (store *BlockStore) RetrieveBlockByNumber(blockNum uint64) (*common.Block,
 	return store.fileMgr.retrieveBlockByNumber(blockNum)
 }
 
+// TxIDExists returns true if a transaction with the txID is ever committed
+func (store *BlockStore) TxIDExists(txID string) (bool, error) {
+	return store.fileMgr.txIDExists(txID)
+}
+
 // RetrieveTxByID returns a transaction for given transaction id
 func (store *BlockStore) RetrieveTxByID(txID string) (*common.Envelope, error) {
 	return store.fileMgr.retrieveTransactionByID(txID)

common/ledger/blkstorage/snapshot_test.go

@@ -491,6 +491,10 @@ func verifyQueriesOnBlocksPriorToSnapshot(
 
 			_, err = bootstrappedBlockStore.RetrieveTxValidationCodeByTxID(txID)
 			require.EqualError(t, err, expectedErrorStr)
+
+			exists, err := bootstrappedBlockStore.TxIDExists(txID)
+			require.NoError(t, err)
+			require.True(t, exists)
 		}
 	}
 }
@@ -540,6 +544,10 @@ func verifyQueriesOnBlocksAddedAfterBootstrapping(t *testing.T,
 			expectedTxEnv, err := protoutil.GetEnvelopeFromBlock(block.Data.Data[j])
 			require.NoError(t, err)
 			require.Equal(t, expectedTxEnv, retrievedTxEnv)
+
+			exists, err := bootstrappedBlockStore.TxIDExists(txID)
+			require.NoError(t, err)
+			require.True(t, exists)
 		}
 
 		for j, validationCode := range d.validationCodes {

core/committer/committer_test.go

@@ -59,6 +59,12 @@ func (m *mockLedger) Close() {
 
 }
 
+// TxIDExists returns true if the specified txID is already present in one of the already committed blocks
+func (m *mockLedger) TxIDExists(txID string) (bool, error) {
+	args := m.Called(txID)
+	return args.Get(0).(bool), args.Error(1)
+}
+
 func (m *mockLedger) GetTransactionByID(txID string) (*peer.ProcessedTransaction, error) {
 	args := m.Called(txID)
 	return args.Get(0).(*peer.ProcessedTransaction), args.Error(1)

core/committer/txvalidator/v14/validator.go

@@ -453,29 +453,25 @@ func (v *TxValidator) checkTxIdDupsLedger(tIdx int, chdr *common.ChannelHeader,
 	txID := chdr.TxId
 
 	// Look for a transaction with the same identifier inside the ledger
-	_, err := ldgr.GetTransactionByID(txID)
+	exists, err := ldgr.TxIDExists(txID)
 
-	switch err.(type) {
-	case nil:
-		// invalid case, returned error is nil. It means that there is already a tx in the ledger with the same id
-		logger.Error("Duplicate transaction found, ", txID, ", skipping")
-		return &blockValidationResult{
-			tIdx:           tIdx,
-			validationCode: peer.TxValidationCode_DUPLICATE_TXID,
-		}
-	case ledger.NotFoundInIndexErr:
-		// valid case, returned error is of type NotFoundInIndexErr.
-		// It means that no tx with the same id is found in the ledger
-		return nil
-	default:
-		// invalid case, returned error is not of type NotFoundInIndexErr.
-		// It means that we could not verify whether a tx with the supplied id is in the ledger
+	if err != nil {
 		logger.Errorf("Ledger failure while attempting to detect duplicate status for txid %s: %s", txID, err)
 		return &blockValidationResult{
 			tIdx: tIdx,
 			err:  err,
 		}
 	}
+
+	if exists {
+		logger.Error("Duplicate transaction found, ", txID, ", skipping")
+		return &blockValidationResult{
+			tIdx:           tIdx,
+			validationCode: peer.TxValidationCode_DUPLICATE_TXID,
+		}
+	}
+
+	return nil
 }
 
 // generateCCKey generates a unique identifier for chaincode in specific channel

core/committer/txvalidator/v14/validator_test.go

@@ -1426,6 +1426,12 @@ type mockLedger struct {
 	mock.Mock
 }
 
+// TxIDExists returns true if the specified txID is already present in one of the already committed blocks
+func (m *mockLedger) TxIDExists(txID string) (bool, error) {
+	args := m.Called(txID)
+	return args.Get(0).(bool), args.Error(1)
+}
+
 // GetTransactionByID returns transaction by ud
 func (m *mockLedger) GetTransactionByID(txID string) (*peer.ProcessedTransaction, error) {
 	args := m.Called(txID)
@@ -1715,7 +1721,7 @@ func TestLedgerIsNoAvailable(t *testing.T) {
 	ccID := "mycc"
 	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
 
-	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, ledger.NotFoundInIndexErr(""))
+	theLedger.On("TxIDExists", mock.Anything).Return(false, nil)
 
 	queryExecutor := new(mockQueryExecutor)
 	queryExecutor.On("GetState", mock.Anything, mock.Anything).Return([]byte{}, errors.New("Unable to connect to DB"))
@@ -1751,7 +1757,7 @@ func TestLedgerIsNotAvailableForCheckingTxidDuplicate(t *testing.T) {
 	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"))
+	theLedger.On("TxIDExists", mock.Anything).Return(false, errors.New("Unable to connect to DB"))
 
 	b := &common.Block{
 		Data:   &common.BlockData{Data: [][]byte{protoutil.MarshalOrPanic(tx)}},
@@ -1781,7 +1787,7 @@ func TestDuplicateTxId(t *testing.T) {
 	ccID := "mycc"
 	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
 
-	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, nil)
+	theLedger.On("TxIDExists", mock.Anything).Return(true, nil)
 
 	b := &common.Block{
 		Data:   &common.BlockData{Data: [][]byte{protoutil.MarshalOrPanic(tx)}},
@@ -1822,7 +1828,7 @@ func TestValidationInvalidEndorsing(t *testing.T) {
 	ccID := "mycc"
 	tx := getEnv(ccID, nil, createRWset(t, ccID), t)
 
-	theLedger.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, ledger.NotFoundInIndexErr(""))
+	theLedger.On("TxIDExists", mock.Anything).Return(false, nil)
 
 	cd := &ccp.ChaincodeData{
 		Name:    ccID,
@@ -1851,7 +1857,7 @@ func TestValidationInvalidEndorsing(t *testing.T) {
 
 func createMockLedger(t *testing.T, ccID string) *mockLedger {
 	l := new(mockLedger)
-	l.On("GetTransactionByID", mock.Anything).Return(&peer.ProcessedTransaction{}, ledger.NotFoundInIndexErr(""))
+	l.On("TxIDExists", mock.Anything).Return(false, nil)
 	cd := &ccp.ChaincodeData{
 		Name:    ccID,
 		Version: ccVersion,