[FAB-2086] API method to get block by TxID

https://jira.hyperledger.org/browse/FAB-2086

We need this method for getting information
of block in a ledger by txID.
Also this method helps to check valid tx or not
by getting the validation bit array
from the block.

Change-Id: I4216635097ecd4cf5de9e3442dfbd5660d44292f
Signed-off-by: Ruslan Kryukov <ruslan.kryukov@ru.ibm.com>

common/ledger/blkstorage/blockstorage.go

@@ -32,6 +32,7 @@ const (
  IndexableAttrBlockHash = IndexableAttr("BlockHash")
  IndexableAttrTxID = IndexableAttr("TxID")
  IndexableAttrBlockNumTranNum = IndexableAttr("BlockNumTranNum")
+ IndexableAttrBlockTxID = IndexableAttr("BlockTxID")
 )
 
 // IndexConfig - a configuration that includes a list of attributes that should be indexed
@@ -66,5 +67,6 @@ type BlockStore interface {
  RetrieveBlockByNumber(blockNum uint64) (*common.Block, error) // blockNum of math.MaxUint64 will return last block
  RetrieveTxByID(txID string) (*common.Envelope, error)
  RetrieveTxByBlockNumTranNum(blockNum uint64, tranNum uint64) (*common.Envelope, error)
+ RetrieveBlockByTxID(txID string) (*common.Block, error)
  Shutdown()
 }

common/ledger/blkstorage/fsblkstorage/blockfile_mgr.go

@@ -438,6 +438,17 @@ func (mgr *blockfileMgr) retrieveBlockByNumber(blockNum uint64) (*common.Block,
  return mgr.fetchBlock(loc)
 }
 
+func (mgr *blockfileMgr) retrieveBlockByTxID(txID string) (*common.Block, error) {
+ logger.Debugf("retrieveBlockByTxID() - txID = [%s]", txID)
+
+ loc, err := mgr.index.getBlockLocByTxID(txID)
+
+ if err != nil {
+ return nil, err
+ }
+ return mgr.fetchBlock(loc)
+}
+
 func (mgr *blockfileMgr) retrieveBlockHeaderByNumber(blockNum uint64) (*common.BlockHeader, error) {
  logger.Debugf("retrieveBlockHeaderByNumber() - blockNum = [%d]", blockNum)
  loc, err := mgr.index.getBlockLocByBlockNum(blockNum)

common/ledger/blkstorage/fsblkstorage/blockfile_mgr_test.go

@@ -216,3 +216,23 @@ func TestBlockfileMgrFileRolling(t *testing.T) {
  testutil.AssertEquals(t, blkfileMgrWrapper.blockfileMgr.cpInfo.latestFileChunkSuffixNum, 2)
  blkfileMgrWrapper.testGetBlockByHash(blocks)
 }
+
+func TestBlockfileMgrGetBlockByTxID(t *testing.T) {
+ env := newTestEnv(t, NewConf(testPath, 0))
+ defer env.Cleanup()
+ blkfileMgrWrapper := newTestBlockfileWrapper(env, "testLedger")
+ defer blkfileMgrWrapper.close()
+ blocks := testutil.ConstructTestBlocks(t, 10)
+ blkfileMgrWrapper.addBlocks(blocks)
+ for _, blk := range blocks {
+ for j, _ := range blk.Data.Data {
+ // blockNum starts with 1
+ txID, err := extractTxID(blk.Data.Data[j])
+ testutil.AssertNoError(t, err, "")
+
+ blockFromFileMgr, err := blkfileMgrWrapper.blockfileMgr.retrieveBlockByTxID(txID)
+ testutil.AssertNoError(t, err, "Error while retrieving block from blkfileMgr")
+ testutil.AssertEquals(t, blockFromFileMgr, blk)
+ }
+ }
+}

common/ledger/blkstorage/fsblkstorage/blockindex.go

@@ -31,6 +31,7 @@ const (
  blockHashIdxKeyPrefix = 'h'
  txIDIdxKeyPrefix = 't'
  blockNumTranNumIdxKeyPrefix = 'a'
+ blockTxIDIdxKeyPrefix = 'b'
  indexCheckpointKeyStr = "indexCheckpointKey"
 )
 
@@ -43,6 +44,7 @@ type index interface {
  getBlockLocByBlockNum(blockNum uint64) (*fileLocPointer, error)
  getTxLoc(txID string) (*fileLocPointer, error)
  getTXLocByBlockNumTranNum(blockNum uint64, tranNum uint64) (*fileLocPointer, error)
+ getBlockLocByTxID(txID string) (*fileLocPointer, error)
 }
 
 type blockIdxInfo struct {
@@ -127,6 +129,13 @@ func (index *blockIndex) indexBlock(blockIdxInfo *blockIdxInfo) error {
  }
  }
 
+ // Index5 - Store BlockNumber will be used to find block by transaction id
+ if _, ok := index.indexItemsMap[blkstorage.IndexableAttrBlockTxID]; ok {
+ for _, txoffset := range txOffsets {
+ batch.Put(constructBlockTxIDKey(txoffset.txID), flpBytes)
+ }
+ }
+
  batch.Put(indexCheckpointKey, encodeBlockNum(blockIdxInfo.blockNum))
  if err := index.db.WriteBatch(batch, false); err != nil {
  return err
@@ -182,6 +191,22 @@ func (index *blockIndex) getTxLoc(txID string) (*fileLocPointer, error) {
  return txFLP, nil
 }
 
+func (index *blockIndex) getBlockLocByTxID(txID string) (*fileLocPointer, error) {
+ if _, ok := index.indexItemsMap[blkstorage.IndexableAttrBlockTxID]; !ok {
+ return nil, blkstorage.ErrAttrNotIndexed
+ }
+ b, err := index.db.Get(constructBlockTxIDKey(txID))
+ if err != nil {
+ return nil, err
+ }
+ if b == nil {
+ return nil, blkstorage.ErrNotFoundInIndex
+ }
+ txFLP := &fileLocPointer{}
+ txFLP.unmarshal(b)
+ return txFLP, nil
+}
+
 func (index *blockIndex) getTXLocByBlockNumTranNum(blockNum uint64, tranNum uint64) (*fileLocPointer, error) {
  if _, ok := index.indexItemsMap[blkstorage.IndexableAttrBlockNumTranNum]; !ok {
  return nil, blkstorage.ErrAttrNotIndexed
@@ -211,6 +236,10 @@ func constructTxIDKey(txID string) []byte {
  return append([]byte{txIDIdxKeyPrefix}, []byte(txID)...)
 }
 
+func constructBlockTxIDKey(txID string) []byte {
+ return append([]byte{blockTxIDIdxKeyPrefix}, []byte(txID)...)
+}
+
 func constructBlockNumTranNumKey(blockNum uint64, txNum uint64) []byte {
  blkNumBytes := util.EncodeOrderPreservingVarUint64(blockNum)
  tranNumBytes := util.EncodeOrderPreservingVarUint64(txNum)

common/ledger/blkstorage/fsblkstorage/blockindex_test.go

@@ -47,6 +47,10 @@ func (i *noopIndex) getTXLocByBlockNumTranNum(blockNum uint64, tranNum uint64) (
  return nil, nil
 }
 
+func (i *noopIndex) getBlockLocByTxID(txID string) (*fileLocPointer, error) {
+ return nil, nil
+}
+
 func TestBlockIndexSync(t *testing.T) {
  testBlockIndexSync(t, 10, 5, false)
  testBlockIndexSync(t, 10, 5, true)
@@ -111,6 +115,7 @@ func TestBlockIndexSelectiveIndexing(t *testing.T) {
  testBlockIndexSelectiveIndexing(t, []blkstorage.IndexableAttr{blkstorage.IndexableAttrBlockNumTranNum})
  testBlockIndexSelectiveIndexing(t, []blkstorage.IndexableAttr{blkstorage.IndexableAttrBlockHash, blkstorage.IndexableAttrBlockNum})
  testBlockIndexSelectiveIndexing(t, []blkstorage.IndexableAttr{blkstorage.IndexableAttrTxID, blkstorage.IndexableAttrBlockNumTranNum})
+ testBlockIndexSelectiveIndexing(t, []blkstorage.IndexableAttr{blkstorage.IndexableAttrBlockTxID})
 }
 
 func testBlockIndexSelectiveIndexing(t *testing.T, indexItems []blkstorage.IndexableAttr) {
@@ -168,4 +173,15 @@ func testBlockIndexSelectiveIndexing(t *testing.T, indexItems []blkstorage.Index
  } else {
  testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
  }
+
+ // test 'retrieveBlockByTxID'
+ txid, err = extractTxID(blocks[0].Data.Data[0])
+ testutil.AssertNoError(t, err, "")
+ block, err = blockfileMgr.retrieveBlockByTxID(txid)
+ if testutil.Contains(indexItems, blkstorage.IndexableAttrBlockTxID) {
+ testutil.AssertNoError(t, err, "Error while retrieving block by txID")
+ testutil.AssertEquals(t, blocks[0], block)
+ } else {
+ testutil.AssertSame(t, err, blkstorage.ErrAttrNotIndexed)
+ }
 }

common/ledger/blkstorage/fsblkstorage/fs_blockstore.go

@@ -77,6 +77,10 @@ func (store *fsBlockStore) RetrieveTxByBlockNumTranNum(blockNum uint64, tranNum
  return store.fileMgr.retrieveTransactionByBlockNumTranNum(blockNum, tranNum)
 }
 
+func (store *fsBlockStore) RetrieveBlockByTxID(txID string) (*common.Block, error) {
+ return store.fileMgr.retrieveBlockByTxID(txID)
+}
+
 // Shutdown shuts down the block store
 func (store *fsBlockStore) Shutdown() {
  logger.Debugf("closing fs blockStore:%s", store.id)

common/ledger/blkstorage/fsblkstorage/pkg_test.go

@@ -41,6 +41,7 @@ func newTestEnv(t testing.TB, conf *Conf) *testEnv {
  blkstorage.IndexableAttrBlockNum,
  blkstorage.IndexableAttrTxID,
  blkstorage.IndexableAttrBlockNumTranNum,
+ blkstorage.IndexableAttrBlockTxID,
  }
  return newTestEnvSelectiveIndexing(t, conf, attrsToIndex)
 }

core/ledger/kvledger/kv_ledger.go

@@ -228,6 +228,11 @@ func (l *kvLedger) GetBlockByHash(blockHash []byte) (*common.Block, error) {
  return l.blockStore.RetrieveBlockByHash(blockHash)
 }
 
+// GetBlockByTxID returns a block which contains a transaction
+func (l *kvLedger) GetBlockByTxID(txID string) (*common.Block, error) {
+ return l.blockStore.RetrieveBlockByTxID(txID)
+}
+
 //Prune prunes the blocks/transactions that satisfy the given policy
 func (l *kvLedger) Prune(policy commonledger.PrunePolicy) error {
  return errors.New("Not yet implemented")

core/ledger/kvledger/kv_ledger_provider.go

@@ -63,6 +63,7 @@ func NewProvider() (ledger.PeerLedgerProvider, error) {
  blkstorage.IndexableAttrBlockNum,
  blkstorage.IndexableAttrTxID,
  blkstorage.IndexableAttrBlockNumTranNum,
+ blkstorage.IndexableAttrBlockTxID,
  }
  indexConfig := &blkstorage.IndexConfig{AttrsToIndex: attrsToIndex}
  blockStoreProvider := fsblkstorage.NewProvider(

core/ledger/ledger_interface.go

@@ -43,6 +43,8 @@ type PeerLedger interface {
  GetTransactionByID(txID string) (*common.Envelope, error)
  // GetBlockByHash returns a block given it's hash
  GetBlockByHash(blockHash []byte) (*common.Block, error)
+ // GetBlockByTxID returns a block which contains a transaction
+ GetBlockByTxID(txID string) (*common.Block, error)
  // NewTxSimulator gives handle to a transaction simulator.
  // A client can obtain more than one 'TxSimulator's for parallel execution.
  // Any snapshoting/synchronization should be performed at the implementation level if required

core/scc/qscc/querier.go

@@ -45,6 +45,7 @@ const (
  GetBlockByNumber string = "GetBlockByNumber"
  GetBlockByHash string = "GetBlockByHash"
  GetTransactionByID string = "GetTransactionByID"
+ GetBlockByTxID string = "GetBlockByTxID"
 )
 
 // Init is called once per chain when the chain is created.
@@ -95,6 +96,8 @@ func (e *LedgerQuerier) Invoke(stub shim.ChaincodeStubInterface) pb.Response {
  return getBlockByHash(targetLedger, args[2])
  case GetChainInfo:
  return getChainInfo(targetLedger)
+ case GetBlockByTxID:
+ return getBlockByTxID(targetLedger, args[2])
  }
 
  return shim.Error(fmt.Sprintf("Requested function %s not found.", fname))
@@ -177,3 +180,20 @@ func getChainInfo(vledger ledger.PeerLedger) pb.Response {
 
  return shim.Success(bytes)
 }
+
+func getBlockByTxID(vledger ledger.PeerLedger, rawTxID []byte) pb.Response {
+ txID := string(rawTxID)
+ block, err := vledger.GetBlockByTxID(txID)
+
+ if err != nil {
+ return shim.Error(fmt.Sprintf("Failed to get block for txID %s, error %s", txID, err))
+ }
+
+ bytes, err := utils.Marshal(block)
+
+ if err != nil {
+ return shim.Error(err.Error())
+ }
+
+ return shim.Success(bytes)
+}

core/scc/qscc/querier_test.go

@@ -117,3 +117,20 @@ func TestQueryGetBlockByHash(t *testing.T) {
  t.Fatalf("qscc GetBlockByHash should have failed with invalid hash: 0")
  }
 }
+
+func TestQueryGetBlockByTxID(t *testing.T) {
+ viper.Set("peer.fileSystemPath", "/var/hyperledger/test8/")
+ defer os.RemoveAll("/var/hyperledger/test8/")
+ peer.MockInitialize()
+ peer.MockCreateChain("mytestchainid8")
+
+ e := new(LedgerQuerier)
+ stub := shim.NewMockStub("LedgerQuerier", e)
+
+ txID := ""
+
+ args := [][]byte{[]byte(GetBlockByTxID), []byte("mytestchainid8"), []byte(txID)}
+ if res := stub.MockInvoke("1", args); res.Status == shim.OK {
+ t.Fatalf("qscc GetBlockByTxID should have failed with invalid txID: %s", txID)
+ }
+}