[FAB-6433] Pass ledger height at Persist()

This change set makes the coordinator pass the ledger height when
it persists private data into the transient store.

Change-Id: Id890b4ce6043b841e0d9c9db7959e01e393b2f78
Signed-off-by: yacovm <yacovm@il.ibm.com>

Author: yacovm

gossip/privdata/coordinator.go

@@ -113,8 +113,11 @@ func NewCoordinator(support Support, selfSignedData common.SignedData) Coordinat
 
 // StorePvtData used to persist private date into transient store
 func (c *coordinator) StorePvtData(txID string, privData *rwset.TxPvtReadWriteSet) error {
-	// TODO pass received at block height instead of 0
-	return c.TransientStore.Persist(txID, 0, privData)
+	height, err := c.Support.LedgerHeight()
+	if err != nil {
+		return errors.Wrap(err, "failed obtaining ledger height, thus cannot persist private data")
+	}
+	return c.TransientStore.Persist(txID, height, privData)
 }
 
 // StoreBlock stores block with private data into the ledger
@@ -239,8 +242,9 @@ func (c *coordinator) fetchFromPeers(blockSeq uint64, ownedRWsets map[rwSetKey][
 			}
 			ownedRWsets[key] = rws
 			delete(privateInfo.missingKeys, key)
-			// TODO Pass received at block height instead of 0
-			c.TransientStore.Persist(dig.TxId, 0, key.toTxPvtReadWriteSet(rws))
+			// If we fetch private data that is associated to block i, then our last block persisted must be i-1
+			// so our ledger height is i, since blocks start from 0.
+			c.TransientStore.Persist(dig.TxId, blockSeq, key.toTxPvtReadWriteSet(rws))
 			logger.Debug("Fetched", key)
 		}
 	}

gossip/privdata/coordinator_test.go

@@ -80,7 +80,7 @@ func (store *mockTransientStore) Persist(txid string, blockHeight uint64, res *r
 		store.t.Fatal("Shouldn't have persisted", res)
 	}
 	delete(store.persists, key)
-	store.Called(txid, res)
+	store.Called(txid, blockHeight, res)
 	return nil
 }
 
@@ -163,6 +163,9 @@ func (mock *committerMock) Commit(block *common.Block) error {
 
 func (mock *committerMock) LedgerHeight() (uint64, error) {
 	args := mock.Called()
+	if args.Get(0) == nil {
+		return uint64(0), args.Error(1)
+	}
 	return args.Get(0).(uint64), args.Error(1)
 }
 
@@ -787,7 +790,7 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 			Payload: [][]byte{[]byte("rws-pre-image")},
 		},
 	}, nil)
-	store.On("Persist", mock.Anything, mock.Anything).
+	store.On("Persist", mock.Anything, uint64(1), mock.Anything).
 		expectRWSet("ns1", "c2", []byte("rws-pre-image")).
 		expectRWSet("ns2", "c1", []byte("rws-pre-image")).Return(nil)
 	pvtData = pdFactory.addRWSet().addNSRWSet("ns1", "c1").create()
@@ -831,7 +834,7 @@ func TestCoordinatorStoreBlock(t *testing.T) {
 		}
 	}).Return(nil)
 	store.On("GetTxPvtRWSetByTxid", "tx3", mock.Anything).Return(&mockRWSetScanner{err: errors.New("uh oh")}, nil)
-	store.On("Persist", mock.Anything, mock.Anything).expectRWSet("ns3", "c3", []byte("rws-pre-image"))
+	store.On("Persist", mock.Anything, uint64(1), mock.Anything).expectRWSet("ns3", "c3", []byte("rws-pre-image"))
 	committer = &committerMock{}
 	committer.On("CommitWithPvtData", mock.Anything).Run(func(args mock.Arguments) {
 		var privateDataPassed2Ledger privateData = args.Get(0).(*ledger.BlockAndPvtData).BlockPvtData
@@ -1032,3 +1035,30 @@ func TestCoordinatorGetBlocks(t *testing.T) {
 	assert.Empty(t, returnedPrivateData)
 	assert.Error(t, err)
 }
+
+func TestCoordinatorStorePvtData(t *testing.T) {
+	cs := createcollectionStore(common.SignedData{}).thatAcceptsAll()
+	committer := &committerMock{}
+	committer.On("LedgerHeight").Return(uint64(5), nil).Once()
+	store := &mockTransientStore{t: t}
+	store.On("Persist", mock.Anything, uint64(5), mock.Anything).
+		expectRWSet("ns1", "c1", []byte("rws-pre-image")).Return(nil)
+	fetcher := &fetcherMock{t: t}
+	coordinator := NewCoordinator(Support{
+		CollectionStore: cs,
+		Committer:       committer,
+		Fetcher:         fetcher,
+		TransientStore:  store,
+		Validator:       &validatorMock{},
+	}, common.SignedData{})
+	pvtData := (&pvtDataFactory{}).addRWSet().addNSRWSet("ns1", "c1").create()
+	// Green path: ledger height can be retrieved from ledger/committer
+	err := coordinator.StorePvtData("tx1", pvtData[0].WriteSet)
+	assert.NoError(t, err)
+
+	// Bad path: ledger height cannot be retrieved from ledger/committer
+	committer.On("LedgerHeight").Return(uint64(0), errors.New("I/O error: file system full"))
+	err = coordinator.StorePvtData("tx1", pvtData[0].WriteSet)
+	assert.Error(t, err)
+	assert.Contains(t, err.Error(), "I/O error: file system full")
+}