[FAB-5854] Add additional unit tests ApplyUpdates()

Add unit tests for additional ApplyUpdates() scenarios:

1) If an individual update in the CouchDB bulk update fails,
we fall back to one-by-one processing of the failed updates with
SaveDoc().
This scenario needs additional unit tests.

Change-Id: I817f6672d5f1336a468e12e6875bf12841a1aaae
Signed-off-by: Chris Elder <chris.elder@us.ibm.com>

Author: Chris Elder

core/ledger/kvledger/txmgmt/statedb/commontests/test_common.go

@@ -614,5 +614,113 @@ func TestSmallBatchSize(t *testing.T, dbProvider statedb.VersionedDBProvider) {
 
 	vv, _ = db.GetState("ns1", "key11")
 	testutil.AssertEquals(t, vv.Value, jsonValue11)
+}
+
+// TestBatchWithIndividualRetry tests a single failure in a batch
+func TestBatchWithIndividualRetry(t *testing.T, dbProvider statedb.VersionedDBProvider) {
+
+	db, err := dbProvider.GetDBHandle("testbatchretry")
+	testutil.AssertNoError(t, err, "")
+
+	batch := statedb.NewUpdateBatch()
+	vv1 := statedb.VersionedValue{Value: []byte("value1"), Version: version.NewHeight(1, 1)}
+	vv2 := statedb.VersionedValue{Value: []byte("value2"), Version: version.NewHeight(1, 2)}
+	vv3 := statedb.VersionedValue{Value: []byte("value3"), Version: version.NewHeight(1, 3)}
+	vv4 := statedb.VersionedValue{Value: []byte("value4"), Version: version.NewHeight(1, 4)}
+
+	batch.Put("ns", "key1", vv1.Value, vv1.Version)
+	batch.Put("ns", "key2", vv2.Value, vv2.Version)
+	batch.Put("ns", "key3", vv3.Value, vv3.Version)
+	batch.Put("ns", "key4", vv4.Value, vv4.Version)
+	savePoint := version.NewHeight(1, 5)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// Clear the cache for the next batch, in place of simulation
+	if bulkdb, ok := db.(statedb.BulkOptimizable); ok {
+		//clear the cached versions, this will force a read when getVerion is called
+		bulkdb.ClearCachedVersions()
+	}
+
+	batch = statedb.NewUpdateBatch()
+	batch.Put("ns", "key1", vv1.Value, vv1.Version)
+	batch.Put("ns", "key2", vv2.Value, vv2.Version)
+	batch.Put("ns", "key3", vv3.Value, vv3.Version)
+	batch.Put("ns", "key4", vv4.Value, vv4.Version)
+	savePoint = version.NewHeight(1, 6)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// Update document key3
+	batch = statedb.NewUpdateBatch()
+	batch.Delete("ns", "key2", vv2.Version)
+	batch.Put("ns", "key3", vv3.Value, vv3.Version)
+	savePoint = version.NewHeight(1, 7)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// This should force a retry for couchdb revision conflict for both delete and update
+	// Retry logic should correct the update and prevent delete from throwing an error
+	batch = statedb.NewUpdateBatch()
+	batch.Delete("ns", "key2", vv2.Version)
+	batch.Put("ns", "key3", vv3.Value, vv3.Version)
+	savePoint = version.NewHeight(1, 8)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	//Create a new set of values that use JSONs instead of binary
+	jsonValue5 := []byte(`{"asset_name": "marble5","color": "blue","size": 5,"owner": "fred"}`)
+	jsonValue6 := []byte(`{"asset_name": "marble6","color": "blue","size": 6,"owner": "elaine"}`)
+	jsonValue7 := []byte(`{"asset_name": "marble7","color": "blue","size": 7,"owner": "fred"}`)
+	jsonValue8 := []byte(`{"asset_name": "marble8","color": "blue","size": 8,"owner": "elaine"}`)
+
+	// Clear the cache for the next batch, in place of simulation
+	if bulkdb, ok := db.(statedb.BulkOptimizable); ok {
+		//clear the cached versions, this will force a read when getVersion is called
+		bulkdb.ClearCachedVersions()
+	}
+
+	batch = statedb.NewUpdateBatch()
+	batch.Put("ns1", "key5", jsonValue5, version.NewHeight(1, 9))
+	batch.Put("ns1", "key6", jsonValue6, version.NewHeight(1, 10))
+	batch.Put("ns1", "key7", jsonValue7, version.NewHeight(1, 11))
+	batch.Put("ns1", "key8", jsonValue8, version.NewHeight(1, 12))
+	savePoint = version.NewHeight(1, 6)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// Clear the cache for the next batch, in place of simulation
+	if bulkdb, ok := db.(statedb.BulkOptimizable); ok {
+		//clear the cached versions, this will force a read when getVersion is called
+		bulkdb.ClearCachedVersions()
+	}
+
+	//Send the batch through again to test updates
+	batch = statedb.NewUpdateBatch()
+	batch.Put("ns1", "key5", jsonValue5, version.NewHeight(1, 9))
+	batch.Put("ns1", "key6", jsonValue6, version.NewHeight(1, 10))
+	batch.Put("ns1", "key7", jsonValue7, version.NewHeight(1, 11))
+	batch.Put("ns1", "key8", jsonValue8, version.NewHeight(1, 12))
+	savePoint = version.NewHeight(1, 6)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// Update document key3
+	// this will cause an inconsistent cache entry for connection db2
+	batch = statedb.NewUpdateBatch()
+	batch.Delete("ns1", "key6", version.NewHeight(1, 13))
+	batch.Put("ns1", "key7", jsonValue7, version.NewHeight(1, 14))
+	savePoint = version.NewHeight(1, 15)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
+
+	// This should force a retry for couchdb revision conflict for both delete and update
+	// Retry logic should correct the update and prevent delete from throwing an error
+	batch = statedb.NewUpdateBatch()
+	batch.Delete("ns1", "key6", version.NewHeight(1, 16))
+	batch.Put("ns1", "key7", jsonValue7, version.NewHeight(1, 17))
+	savePoint = version.NewHeight(1, 18)
+	err = db.ApplyUpdates(batch, savePoint)
+	testutil.AssertNoError(t, err, "")
 
 }

core/ledger/kvledger/txmgmt/statedb/statecouchdb/statecouchdb.go

@@ -34,7 +34,7 @@ var binaryWrapper = "valueBytes"
 // currently defaulted to 0 and is not used
 var querySkip = 0
 
-// BatchDocument defines a document for a batch
+//BatchableDocument defines a document for a batch
 type BatchableDocument struct {
 	CouchDoc couchdb.CouchDoc
 	Deleted  bool
@@ -158,7 +158,7 @@ func (vdb *VersionedDB) GetState(namespace string, key string) (*statedb.Version
 	return &statedb.VersionedValue{Value: returnValue, Version: returnVersion}, nil
 }
 
-// GetVersion implements method in VersionedDB interface
+//GetCachedVersion implements method in VersionedDB interface
 func (vdb *VersionedDB) GetCachedVersion(namespace string, key string) (*version.Height, bool) {
 
 	logger.Debugf("Retrieving cached version: %s~%s", key, namespace)
@@ -214,9 +214,6 @@ func removeDataWrapper(wrappedValue []byte, attachments []*couchdb.AttachmentInf
 	// initialize the return value
 	returnValue := []byte{}
 
-	// initialize a default return version
-	returnVersion := version.NewHeight(0, 0)
-
 	// create a generic map for the json
 	jsonResult := make(map[string]interface{})
 
@@ -242,7 +239,7 @@ func removeDataWrapper(wrappedValue []byte, attachments []*couchdb.AttachmentInf
 
 	}
 
-	returnVersion = createVersionHeightFromVersionString(jsonResult["version"].(string))
+	returnVersion := createVersionHeightFromVersionString(jsonResult["version"].(string))
 
 	return returnValue, returnVersion
 
@@ -417,7 +414,7 @@ func (vdb *VersionedDB) processUpdateBatch(updateBatch *statedb.UpdateBatch, mis
 
 	// Use the batchUpdateMap for tracking couchdb updates by ID
 	// this will be used in case there are retries required
-	batchUpdateMap := make(map[string]interface{})
+	batchUpdateMap := make(map[string]*BatchableDocument)
 
 	namespaces := updateBatch.GetUpdatedNamespaces()
 	for _, ns := range namespaces {
@@ -465,8 +462,8 @@ func (vdb *VersionedDB) processUpdateBatch(updateBatch *statedb.UpdateBatch, mis
 				}
 			}
 
-			// Add the current docment to the update map
-			batchUpdateMap[string(compositeKey)] = BatchableDocument{CouchDoc: *couchDoc, Deleted: isDelete}
+			// Add the current docment, revision and delete flag to the update map
+			batchUpdateMap[string(compositeKey)] = &BatchableDocument{CouchDoc: *couchDoc, Deleted: isDelete}
 
 		}
 	}
@@ -476,7 +473,7 @@ func (vdb *VersionedDB) processUpdateBatch(updateBatch *statedb.UpdateBatch, mis
 		//Add the documents to the batch update array
 		batchUpdateDocs := []*couchdb.CouchDoc{}
 		for _, updateDocument := range batchUpdateMap {
-			batchUpdateDocument := updateDocument.(BatchableDocument)
+			batchUpdateDocument := updateDocument
 			batchUpdateDocs = append(batchUpdateDocs, &batchUpdateDocument.CouchDoc)
 		}
 
@@ -495,17 +492,35 @@ func (vdb *VersionedDB) processUpdateBatch(updateBatch *statedb.UpdateBatch, mis
 			// If the document returned an error, retry the individual document
 			if respDoc.Ok != true {
 
-				batchUpdateDocument := batchUpdateMap[respDoc.ID].(BatchableDocument)
+				batchUpdateDocument := batchUpdateMap[respDoc.ID]
 
 				var err error
 
+				//Remove the "_rev" from the JSON before saving
+				//this will allow the CouchDB retry logic to retry revisions without encountering
+				//a mismatch between the "If-Match" and the "_rev" tag in the JSON
+				if batchUpdateDocument.CouchDoc.JSONValue != nil {
+					err = removeJSONRevision(&batchUpdateDocument.CouchDoc.JSONValue)
+					if err != nil {
+						return err
+					}
+				}
+
 				// Check to see if the document was added to the batch as a delete type document
 				if batchUpdateDocument.Deleted {
+
+					//Log the warning message that a retry is being attempted for batch delete issue
+					logger.Warningf("CouchDB batch document delete encountered an problem. Retrying delete for document ID:%s", respDoc.ID)
+
 					// If this is a deleted document, then retry the delete
 					// If the delete fails due to a document not being found (404 error),
 					// the document has already been deleted and the DeleteDoc will not return an error
 					err = vdb.db.DeleteDoc(respDoc.ID, "")
 				} else {
+
+					//Log the warning message that a retry is being attempted for batch update issue
+					logger.Warningf("CouchDB batch document update encountered an problem. Retrying update for document ID:%s", respDoc.ID)
+
 					// Save the individual document to couchdb
 					// Note that this will do retries as needed
 					_, err = vdb.db.SaveDoc(respDoc.ID, "", &batchUpdateDocument.CouchDoc)
@@ -674,6 +689,32 @@ func createCouchdbDocJSON(id, revision string, value []byte, chaincodeID string,
 	return documentJSON
 }
 
+// removeJSONRevision removes the "_rev" if this is a JSON
+func removeJSONRevision(jsonValue *[]byte) error {
+
+	jsonMap := make(map[string]interface{})
+
+	//Unmarshal the value into a map
+	err := json.Unmarshal(*jsonValue, &jsonMap)
+	if err != nil {
+		logger.Errorf("Failed to unmarshal couchdb JSON data %s\n", err.Error())
+		return err
+	}
+
+	//delete the "_rev" entry
+	delete(jsonMap, "_rev")
+
+	//marshal the updated map back into the byte array
+	*jsonValue, err = json.Marshal(jsonMap)
+	if err != nil {
+		logger.Errorf("Failed to marshal couchdb JSON data %s\n", err.Error())
+		return err
+	}
+
+	return nil
+
+}
+
 // Savepoint docid (key) for couchdb
 const savepointDocID = "statedb_savepoint"