recon: stateDB recovery after a peer crash

Once the pvtData of old blocks are committed to pvtStore,
we need to check whether the stateDB needs to be updated.
If so, we should commit the pvtData to the stateDB.

As a peer can fail after committing pvtData of old blocks,
we might end up with a stateDB which is out of sync from
the pvtStore. Hence, this CR implements a recovery mechanism.

FAB-11806 #done

Change-Id: If6f8aa608080d7e02fb6c4236079814e3ed5c11f
Signed-off-by: senthil <cendhu@gmail.com>
Signed-off-by: manish <manish.sethi@gmail.com>

Author: cendhu

core/ledger/kvledger/kv_ledger.go

@@ -94,6 +94,16 @@ func (l *kvLedger) initBlockStore(btlPolicy pvtdatapolicy.BTLPolicy) {
 //by recommitting last valid blocks
 func (l *kvLedger) recoverDBs() error {
 	logger.Debugf("Entering recoverDB()")
+	if err := l.syncStateAndHistoryDBWithBlockstore(); err != nil {
+		return err
+	}
+	if err := l.syncStateDBWithPvtdatastore(); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (l *kvLedger) syncStateAndHistoryDBWithBlockstore() error {
 	//If there is no block in blockstorage, nothing to recover.
 	info, _ := l.blockStore.GetBlockchainInfo()
 	if info.Height == 0 {
@@ -136,6 +146,29 @@ func (l *kvLedger) recoverDBs() error {
 		recoverers[0].recoverable, recoverers[1].recoverable)
 }
 
+func (l *kvLedger) syncStateDBWithPvtdatastore() error {
+	// TODO: So far, the design philosophy was that the scope of block storage is
+	// limited to storing and retrieving blocks data with certain guarantees and statedb is
+	// for the state management. The higher layer, 'kvledger', coordinates the acts between
+	// the two. However, with maintaining the state of the consumption of blocks (i.e,
+	// lastUpdatedOldBlockList for pvtstore reconciliation) within private data block storage
+	// breaks that assumption. The knowledge of what blocks have been consumed for the purpose
+	// of state update should not lie with the source (i.e., pvtdatastorage). A potential fix
+	// is mentioned in FAB-12731
+	blocksPvtData, err := l.blockStore.GetLastUpdatedOldBlocksPvtData()
+	if err != nil {
+		return err
+	}
+	if err := l.txtmgmt.RemoveStaleAndCommitPvtDataOfOldBlocks(blocksPvtData); err != nil {
+		return err
+	}
+	if err := l.blockStore.ResetLastUpdatedOldBlocksList(); err != nil {
+		return err
+	}
+
+	return nil
+}
+
 //recommitLostBlocks retrieves blocks in specified range and commit the write set to either
 //state DB or history DB or both
 func (l *kvLedger) recommitLostBlocks(firstBlockNum uint64, lastBlockNum uint64, recoverables ...recoverable) error {

core/ledger/kvledger/kv_ledger_test.go

@@ -15,6 +15,7 @@ import (
 	"github.com/hyperledger/fabric/common/ledger/testutil"
 	"github.com/hyperledger/fabric/common/util"
 	lgr "github.com/hyperledger/fabric/core/ledger"
+	"github.com/hyperledger/fabric/core/ledger/kvledger/txmgmt/txmgr"
 	"github.com/hyperledger/fabric/core/ledger/ledgerconfig"
 	ledgertestutil "github.com/hyperledger/fabric/core/ledger/testutil"
 	"github.com/hyperledger/fabric/protos/common"
@@ -185,6 +186,11 @@ func TestKVLedgerBlockStorageWithPvtdata(t *testing.T) {
 }
 
 func TestKVLedgerDBRecovery(t *testing.T) {
+	testSyncStateAndHistoryDBWithBlockstore(t)
+	testSyncStateDBWithPvtdatastore(t)
+}
+
+func testSyncStateAndHistoryDBWithBlockstore(t *testing.T) {
 	env := newTestEnv(t)
 	defer env.cleanup()
 	provider := testutilNewProviderWithCollectionConfig(t,
@@ -366,6 +372,67 @@ func TestKVLedgerDBRecovery(t *testing.T) {
 	)
 }
 
+func testSyncStateDBWithPvtdatastore(t *testing.T) {
+	env := newTestEnv(t)
+	defer env.cleanup()
+	provider := testutilNewProviderWithCollectionConfig(t,
+		"ns", map[string]uint64{"coll": 0},
+	)
+	defer provider.Close()
+	testLedgerid := "testLedger"
+	bg, gb := testutil.NewBlockGenerator(t, testLedgerid, false)
+	ledger, _ := provider.Create(gb)
+	defer ledger.Close()
+
+	// create and commit two data block (both with missing pvtdata)
+	blockAndPvtdata1, pvtdata1 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk1",
+		map[string]string{"key1": "value1.1", "key2": "value2.1", "key3": "value3.1"},
+		map[string]string{"key1": "pvtValue1.1", "key2": "pvtValue2.1", "key3": "pvtValue3.1"})
+
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata1))
+
+	blockAndPvtdata2, pvtdata2 := prepareNextBlockWithMissingPvtDataForTest(t, ledger, bg, "SimulateForBlk2",
+		map[string]string{"key1": "value1.2", "key2": "value2.2", "key3": "value3.2"},
+		map[string]string{"key1": "pvtValue1.2", "key2": "pvtValue2.2", "key3": "pvtValue3.2"})
+
+	assert.NoError(t, ledger.CommitWithPvtData(blockAndPvtdata2))
+
+	txSim, err := ledger.NewTxSimulator("test")
+	assert.NoError(t, err)
+	value, err := txSim.GetPrivateData("ns", "coll", "key1")
+	_, ok := err.(*txmgr.ErrPvtdataNotAvailable)
+	assert.True(t, ok)
+	assert.Nil(t, value)
+
+	blocksPvtData := map[uint64][]*lgr.TxPvtData{
+		1: {
+			pvtdata1,
+		},
+		2: {
+			pvtdata2,
+		},
+	}
+
+	assert.NoError(t, ledger.(*kvLedger).blockStore.CommitPvtDataOfOldBlocks(blocksPvtData))
+
+	// Now, assume that peer fails here before committing the pvtData to stateDB
+	ledger.Close()
+	provider.Close()
+
+	// Here the peer comes online and calls NewKVLedger to get a handler for the ledger
+	// StateDB and HistoryDB should be recovered before returning from NewKVLedger call
+	provider = testutilNewProviderWithCollectionConfig(t,
+		"ns", map[string]uint64{"coll": 0},
+	)
+	ledger, _ = provider.Open(testLedgerid)
+
+	txSim, err = ledger.NewTxSimulator("test")
+	assert.NoError(t, err)
+	value, err = txSim.GetPrivateData("ns", "coll", "key1")
+	assert.NoError(t, err)
+	assert.Equal(t, value, []byte("pvtValue1.2"))
+}
+
 func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 
 	//call a helper method to load the core.yaml
@@ -482,6 +549,21 @@ func TestLedgerWithCouchDbEnabledWithBinaryAndJSONData(t *testing.T) {
 	}
 }
 
+func prepareNextBlockWithMissingPvtDataForTest(t *testing.T, l lgr.PeerLedger, bg *testutil.BlockGenerator,
+	txid string, pubKVs map[string]string, pvtKVs map[string]string) (*lgr.BlockAndPvtData, *lgr.TxPvtData) {
+
+	blockAndPvtData := prepareNextBlockForTest(t, l, bg, txid, pubKVs, pvtKVs)
+
+	blkMissingDataInfo := make(lgr.TxMissingPvtDataMap)
+	blkMissingDataInfo.Add(0, "ns", "coll", true)
+	blockAndPvtData.MissingPvtData = blkMissingDataInfo
+
+	pvtData := blockAndPvtData.PvtData[0]
+	delete(blockAndPvtData.PvtData, 0)
+
+	return blockAndPvtData, pvtData
+}
+
 func prepareNextBlockForTest(t *testing.T, l lgr.PeerLedger, bg *testutil.BlockGenerator,
 	txid string, pubKVs map[string]string, pvtKVs map[string]string) *lgr.BlockAndPvtData {
 	simulator, _ := l.NewTxSimulator(txid)

core/ledger/kvledger/txmgmt/txmgr/lockbasedtxmgr/lockbased_txmgr.go

@@ -138,7 +138,7 @@ func (txmgr *LockBasedTxMgr) RemoveStaleAndCommitPvtDataOfOldBlocks(blocksPvtDat
 	// a given <ns, coll, key>, we need to find duplicate tuples with different
 	// versions and use the one with the higher version
 	uniquePvtData, err := constructUniquePvtData(blocksPvtData)
-	if err != nil {
+	if len(uniquePvtData) == 0 || err != nil {
 		return err
 	}
 

core/ledger/ledger_interface.go

@@ -253,7 +253,7 @@ type BlockAndPvtData struct {
 // BlockPvtData contains the private data for a block
 type BlockPvtData struct {
 	BlockNum  uint64
-	WriteSets map[uint64]*TxPvtData
+	WriteSets TxPvtDataMap
 }
 
 // Add adds a given missing private data in the MissingPrivateDataList

core/ledger/ledgerstorage/store.go

@@ -231,9 +231,9 @@ func (s *Store) ProcessCollsEligibilityEnabled(committingBlk uint64, nsCollMap m
 	return s.pvtdataStore.ProcessCollsEligibilityEnabled(committingBlk, nsCollMap)
 }
 
-// GetLastUpdatedOldBlocksList invokes the function on underlying pvtdata store
-func (s *Store) GetLastUpdatedOldBlocksList() ([]uint64, error) {
-	return s.pvtdataStore.GetLastUpdatedOldBlocksList()
+// GetLastUpdatedOldBlocksPvtData invokes the function on underlying pvtdata store
+func (s *Store) GetLastUpdatedOldBlocksPvtData() (map[uint64][]*ledger.TxPvtData, error) {
+	return s.pvtdataStore.GetLastUpdatedOldBlocksPvtData()
 }
 
 // ResetLastUpdatedOldBlocksList invokes the function on underlying pvtdata store

core/ledger/pvtdatastorage/store.go

@@ -72,8 +72,8 @@ type Store interface {
 	// these pvtData, the `lastUpdatedOldBlocksList` must be removed. During the peer startup,
 	// if the `lastUpdatedOldBlocksList` exists, stateDB needs to be updated with the appropriate pvtData.
 	CommitPvtDataOfOldBlocks(blocksPvtData map[uint64][]*ledger.TxPvtData) error
-	// GetLastUpdatedOldBlocksList returns the value of `lastUpdatedOldBlocksList`
-	GetLastUpdatedOldBlocksList() ([]uint64, error)
+	// GetLastUpdatedOldBlocksPvtData returns the pvtdata of blocks listed in `lastUpdatedOldBlocksList`
+	GetLastUpdatedOldBlocksPvtData() (map[uint64][]*ledger.TxPvtData, error)
 	// ResetLastUpdatedOldBlocksList removes the `lastUpdatedOldBlocksList` entry from the store
 	ResetLastUpdatedOldBlocksList() error
 	// IsEmpty returns true if the store does not have any block committed yet

core/ledger/pvtdatastorage/store_impl.go

@@ -147,7 +147,7 @@ func (s *store) initState() error {
 	if s.batchPending, err = s.hasPendingCommit(); err != nil {
 		return err
 	}
-	if blist, err = s.GetLastUpdatedOldBlocksList(); err != nil {
+	if blist, err = s.getLastUpdatedOldBlocksList(); err != nil {
 		return err
 	}
 	if len(blist) > 0 {
@@ -521,7 +521,27 @@ func (s *store) commitBatch(batch *leveldbhelper.UpdateBatch) error {
 	return nil
 }
 
-func (s *store) GetLastUpdatedOldBlocksList() ([]uint64, error) {
+// GetLastUpdatedOldBlocksPvtData implements the function in the interface `Store`
+func (s *store) GetLastUpdatedOldBlocksPvtData() (map[uint64][]*ledger.TxPvtData, error) {
+	if !s.isLastUpdatedOldBlocksSet {
+		return nil, nil
+	}
+
+	updatedBlksList, err := s.getLastUpdatedOldBlocksList()
+	if err != nil {
+		return nil, err
+	}
+
+	blksPvtData := make(map[uint64][]*ledger.TxPvtData)
+	for _, blkNum := range updatedBlksList {
+		if blksPvtData[blkNum], err = s.GetPvtDataByBlockNum(blkNum, nil); err != nil {
+			return nil, err
+		}
+	}
+	return blksPvtData, nil
+}
+
+func (s *store) getLastUpdatedOldBlocksList() ([]uint64, error) {
 	var v []byte
 	var err error
 	if v, err = s.db.Get(lastUpdatedOldBlocksKey); err != nil {
@@ -547,10 +567,8 @@ func (s *store) GetLastUpdatedOldBlocksList() ([]uint64, error) {
 	return updatedBlksList, nil
 }
 
+// ResetLastUpdatedOldBlocksList implements the function in the interface `Store`
 func (s *store) ResetLastUpdatedOldBlocksList() error {
-	if !s.isLastUpdatedOldBlocksSet {
-		return &ErrIllegalCall{"No updated old block list"}
-	}
 	batch := leveldbhelper.NewUpdateBatch()
 	batch.Delete(lastUpdatedOldBlocksKey)
 	if err := s.db.WriteBatch(batch, true); err != nil {
@@ -712,6 +730,7 @@ func (s *store) GetMissingPvtDataInfoForMostRecentBlocks(maxBlock int) (ledger.M
 	return missingPvtDataInfo, nil
 }
 
+// ProcessCollsEligibilityEnabled implements the function in the interface `Store`
 func (s *store) ProcessCollsEligibilityEnabled(committingBlk uint64, nsCollMap map[string][]string) error {
 	key := encodeCollElgKey(committingBlk)
 	m := newCollElgInfo(nsCollMap)

core/ledger/pvtdatastorage/store_impl_test.go

@@ -246,7 +246,7 @@ func TestCommitPvtDataOfOldBlocks(t *testing.T) {
 	assert.NoError(err)
 	assert.Equal(expectedMissingPvtDataInfo, missingPvtDataInfo)
 
-	// COMMIT THE MISSINGDATA IN BLOCK 1
+	// COMMIT THE MISSINGDATA IN BLOCK 1 AND BLOCK 2
 	oldBlocksPvtData := make(map[uint64][]*ledger.TxPvtData)
 	oldBlocksPvtData[1] = []*ledger.TxPvtData{
 		produceSamplePvtdata(t, 1, []string{"ns-1:coll-1", "ns-2:coll-1"}),
@@ -298,18 +298,28 @@ func TestCommitPvtDataOfOldBlocks(t *testing.T) {
 	assert.NoError(err)
 	assert.Equal(expectedMissingPvtDataInfo, missingPvtDataInfo)
 
-	expectedBlockList := []uint64{1, 2}
-	blockList, err := store.GetLastUpdatedOldBlocksList()
+	// blksPvtData returns all the pvt data for a block for which the any pvtdata has been submitted
+	// using CommitPvtDataOfOldBlocks
+	blksPvtData, err := store.GetLastUpdatedOldBlocksPvtData()
 	assert.NoError(err)
-	assert.Equal(expectedBlockList, blockList)
+
+	expectedLastupdatedPvtdata := make(map[uint64][]*ledger.TxPvtData)
+	expectedLastupdatedPvtdata[1] = []*ledger.TxPvtData{
+		produceSamplePvtdata(t, 1, []string{"ns-1:coll-1", "ns-2:coll-1"}),
+		produceSamplePvtdata(t, 2, []string{"ns-1:coll-1", "ns-2:coll-1", "ns-2:coll-2", "ns-3:coll-1"}),
+		produceSamplePvtdata(t, 4, []string{"ns-1:coll-1", "ns-1:coll-2", "ns-2:coll-1", "ns-2:coll-2"}),
+	}
+	expectedLastupdatedPvtdata[2] = []*ledger.TxPvtData{
+		produceSamplePvtdata(t, 3, []string{"ns-1:coll-1"}),
+	}
+	assert.Equal(expectedLastupdatedPvtdata, blksPvtData)
 
 	err = store.ResetLastUpdatedOldBlocksList()
 	assert.NoError(err)
 
-	expectedBlockList = []uint64(nil)
-	blockList, err = store.GetLastUpdatedOldBlocksList()
+	blksPvtData, err = store.GetLastUpdatedOldBlocksPvtData()
 	assert.NoError(err)
-	assert.Nil(blockList)
+	assert.Nil(blksPvtData)
 
 	// COMMIT BLOCK 3 WITH NO PVTDATA
 	assert.NoError(store.Prepare(3, nil, nil))