CBG-4790: reconcile rev tree on write for ISGR

Author: gregns1

db/blip_handler.go

@@ -1093,9 +1093,6 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
 	var incomingHLV *HybridLogicalVector
 	// Build history/HLV
 	var legacyRevList []string
-	// we can probably use legacyRevList instead of this but to avoid hooking this up to write code we will use
-	// separate list for now, pending CBG-4790
-	var revTreeProperty []string
 	changeIsVector := strings.Contains(rev, "@")
 	if !bh.useHLV() || !changeIsVector {
 		newDoc.RevID = rev
@@ -1121,11 +1118,13 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
 		newDoc.HLV = incomingHLV
 	}
 
+	isBlipRevTreeProperty := false
 	// if the client is SGW and there are no legacy revs being sent (i.e. doc is not a pre-upgraded doc) check the rev tree property
 	if bh.clientType == BLIPClientTypeSGR2 && len(legacyRevList) == 0 {
 		revTree, ok := rq.Properties[RevMessageTreeHistory]
 		if ok {
-			revTreeProperty = append(revTreeProperty, strings.Split(revTree, ",")...) // nolint: all
+			legacyRevList = append(legacyRevList, strings.Split(revTree, ",")...)
+			isBlipRevTreeProperty = true
 		}
 	}
 
@@ -1349,7 +1348,7 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
 	// bh.conflictResolver != nil represents an active SGR2 and BLIPClientTypeSGR2 represents a passive SGR2
 	forceAllowConflictingTombstone := newDoc.Deleted && (bh.conflictResolver != nil || bh.clientType == BLIPClientTypeSGR2)
 	if bh.useHLV() && changeIsVector {
-		_, _, _, err = bh.collection.PutExistingCurrentVersion(bh.loggingCtx, newDoc, incomingHLV, rawBucketDoc, legacyRevList)
+		_, _, _, err = bh.collection.PutExistingCurrentVersion(bh.loggingCtx, newDoc, incomingHLV, rawBucketDoc, legacyRevList, isBlipRevTreeProperty)
 	} else if bh.conflictResolver != nil {
 		_, _, err = bh.collection.PutExistingRevWithConflictResolution(bh.loggingCtx, newDoc, history, true, bh.conflictResolver, forceAllowConflictingTombstone, rawBucketDoc, ExistingVersionWithUpdateToHLV)
 	} else {

db/blip_sync_context.go

@@ -372,6 +372,7 @@ func (bsc *BlipSyncContext) handleChangesResponse(ctx context.Context, sender *b
 			if bsc.useDeltas && len(knownRevsArray) > 0 {
 				if revID, ok := knownRevsArray[0].(string); ok {
 					if versionVectorProtocol {
+						// extract cv from the known revs array
 						msgHLV, _, err := ExtractHLVFromBlipMessage(revID)
 						if err != nil {
 							base.DebugfCtx(ctx, base.KeySync, "Invalid known rev format for hlv on doc: %s falling back to full body replication.", base.UD(docID))
@@ -385,17 +386,28 @@ func (bsc *BlipSyncContext) handleChangesResponse(ctx context.Context, sender *b
 				}
 			}
 
+			var cvInKnownRevs bool
+
 			for _, rev := range knownRevsArray {
 				if revID, ok := rev.(string); ok {
+					// extract cv from the known revs array
 					msgHLV, _, err := ExtractHLVFromBlipMessage(revID)
 					if err != nil {
-						// assume we have received legacy rev if we cannot parse hlv from known revs, and we are in vv replication
+						// assume we have received legacy rev if the following conditions are met:
+						//  - we cannot parse cv from known revs
+						//  - we are in vv enabled replication
+						// 	- the first element of known rev array was NOT a CV
 						if versionVectorProtocol {
-							legacyRev = true
+							if !cvInKnownRevs {
+								// if cv wasn't the first element in known revs array, this will be legacy doc
+								legacyRev = true
+							}
 						}
 					} else {
 						// extract cv as string
 						revID = msgHLV.GetCurrentVersionString()
+						// mark that we have seen a CV in known revs so we can mark as non legacy rev
+						cvInKnownRevs = true
 					}
 					// we can assume here that if we fail to parse hlv, we have received a rev id in known revs. If we don't fail to parse hlv
 					// then we have extracted cv from it and can assign the cv string to known revs here

db/crud.go

@@ -1211,7 +1211,7 @@ func (db *DatabaseCollectionWithUser) Put(ctx context.Context, docid string, bod
 	return newRevID, doc, err
 }
 
-func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Context, newDoc *Document, newDocHLV *HybridLogicalVector, existingDoc *sgbucket.BucketDocument, revTreeHistory []string) (doc *Document, cv *Version, newRevID string, err error) {
+func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Context, newDoc *Document, newDocHLV *HybridLogicalVector, existingDoc *sgbucket.BucketDocument, revTreeHistory []string, revTreeISGRProperty bool) (doc *Document, cv *Version, newRevID string, err error) {
 	var matchRev string
 	if existingDoc != nil {
 		doc, unmarshalErr := db.unmarshalDocumentWithXattrs(ctx, newDoc.ID, existingDoc.Body, existingDoc.Xattrs, existingDoc.Cas, DocUnmarshalRev)
@@ -1262,7 +1262,14 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 		} else {
 			previousRevTreeID = revTreeHistory[0]
 			prevGeneration, _ = ParseRevID(ctx, previousRevTreeID)
-			newGeneration = prevGeneration + 1
+			// if incoming rev tree list is from a legacy pre upgraded doc, we should have new revID generation based
+			// off the previous current rev +1. If we have rev tree list filled from ISGR's rev tree property then we
+			// should use the current rev of inc
+			if !revTreeISGRProperty {
+				newGeneration = prevGeneration + 1
+			} else {
+				newGeneration = prevGeneration
+			}
 		}
 		revTreeConflictChecked := false
 		var parent string
@@ -1276,6 +1283,12 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 			if addNewerVersionsErr != nil {
 				return nil, nil, false, nil, addNewerVersionsErr
 			}
+			if revTreeISGRProperty {
+				err = doc.alignRevTreeHistory(ctx, newDoc, revTreeHistory)
+				if err != nil {
+					return nil, nil, false, nil, err
+				}
+			}
 		} else {
 			if doc.HLV.isDominating(newDocHLV) {
 				base.DebugfCtx(ctx, base.KeyCRUD, "PutExistingCurrentVersion(%q): No new versions to add.  existing: %#v  new:%#v", base.UD(newDoc.ID), doc.HLV, newDocHLV)
@@ -1289,15 +1302,28 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 				}
 				// the new document has a dominating hlv, so we can ignore any legacy rev revtree information on the incoming document
 				revTreeConflictChecked = true
-				previousRevTreeID = doc.CurrentRev
+				if !revTreeISGRProperty {
+					previousRevTreeID = doc.CurrentRev
+				} else {
+					// align rev tree here for ISGR replications
+					err = doc.alignRevTreeHistory(ctx, newDoc, revTreeHistory)
+					if err != nil {
+						return nil, nil, false, nil, err
+					}
+				}
 			} else {
-				if len(revTreeHistory) > 0 {
+				// if the legacy rev list is from ISGR property, we should not do a conflict check
+				if len(revTreeHistory) > 0 && !revTreeISGRProperty {
 					// conflict check on rev tree history, if there is a rev in rev tree history we have the parent of locally we are not in conflict
 					parent, currentRevIndex, err = db.revTreeConflictCheck(ctx, revTreeHistory, doc, newDoc.Deleted)
 					if err != nil {
 						base.InfofCtx(ctx, base.KeyCRUD, "conflict detected between the two HLV's for doc %s, incoming version %s, local version %s, and conflict found in rev tree history", base.UD(doc.ID), newDocHLV.GetCurrentVersionString(), doc.HLV.GetCurrentVersionString())
 						return nil, nil, false, nil, err
 					}
+					err = doc.addNewerRevisionsToHistory(newDoc, currentRevIndex, parent, revTreeHistory)
+					if err != nil {
+						return nil, nil, false, nil, err
+					}
 					revTreeConflictChecked = true
 					addNewerVersionsErr := doc.HLV.AddNewerVersions(newDocHLV)
 					if addNewerVersionsErr != nil {
@@ -1317,24 +1343,15 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 		// rev tree conflict check if we have rev tree history to check against + finds current rev index to allow us
 		// to add any new revision to rev tree below.
 		// Only check for rev tree conflicts if we haven't already checked above
-		if !revTreeConflictChecked && len(revTreeHistory) > 0 {
+		if !revTreeConflictChecked && len(revTreeHistory) > 0 && !revTreeISGRProperty {
 			parent, currentRevIndex, err = db.revTreeConflictCheck(ctx, revTreeHistory, doc, newDoc.Deleted)
 			if err != nil {
 				return nil, nil, false, nil, err
 			}
-		}
-		// Add all the new revisions to the rev tree:
-		for i := currentRevIndex - 1; i >= 0; i-- {
-			err := doc.History.addRevision(newDoc.ID,
-				RevInfo{
-					ID:      revTreeHistory[i],
-					Parent:  parent,
-					Deleted: i == 0 && newDoc.Deleted})
-
+			err = doc.addNewerRevisionsToHistory(newDoc, currentRevIndex, parent, revTreeHistory)
 			if err != nil {
 				return nil, nil, false, nil, err
 			}
-			parent = revTreeHistory[i]
 		}
 
 		// Process the attachments, replacing bodies with digests.
@@ -1344,16 +1361,23 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 		}
 
 		// generate rev id for new arriving doc
-		strippedBody, _ := stripInternalProperties(newDoc._body)
-		encoding, err := base.JSONMarshalCanonical(strippedBody)
-		if err != nil {
-			return nil, nil, false, nil, err
-		}
-		newRev := CreateRevIDWithBytes(newGeneration, previousRevTreeID, encoding)
-
-		if err := doc.History.addRevision(newDoc.ID, RevInfo{ID: newRev, Parent: previousRevTreeID, Deleted: newDoc.Deleted}); err != nil {
-			base.InfofCtx(ctx, base.KeyCRUD, "Failed to add revision ID: %s, for doc: %s, error: %v", newRev, base.UD(newDoc.ID), err)
-			return nil, nil, false, nil, base.ErrRevTreeAddRevFailure
+		var newRev string
+		if !revTreeISGRProperty {
+			// create a new revID for incoming write
+			strippedBody, _ := stripInternalProperties(newDoc._body)
+			encoding, err := base.JSONMarshalCanonical(strippedBody)
+			if err != nil {
+				return nil, nil, false, nil, err
+			}
+			newRev = CreateRevIDWithBytes(newGeneration, previousRevTreeID, encoding)
+			if err := doc.History.addRevision(newDoc.ID, RevInfo{ID: newRev, Parent: previousRevTreeID, Deleted: newDoc.Deleted}); err != nil {
+				base.InfofCtx(ctx, base.KeyCRUD, "Failed to add revision ID: %s, for doc: %s, error: %v", newRev, base.UD(newDoc.ID), err)
+				return nil, nil, false, nil, base.ErrRevTreeAddRevFailure
+			}
+		} else {
+			// for ISGR, incoming writes current rev should be the most recent rev in history. This aligns the
+			// rev history each of the replication
+			newRev = previousRevTreeID
 		}
 
 		newDoc.RevID = newRev
@@ -3137,7 +3161,6 @@ func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, do
 	if strings.HasPrefix(docid, "_design/") && db.user != nil {
 		return // Users can't upload design docs, so ignore them
 	}
-	// todo: CBG-4782 utilise known revs for rev tree property in ISGR by returning know rev tree id's in possible list
 
 	doc, err := db.GetDocSyncDataNoImport(ctx, docid, DocUnmarshalSync)
 	if err != nil {
@@ -3147,6 +3170,14 @@ func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, do
 		missing = append(missing, rev)
 		return
 	}
+	if doc.HLV == nil {
+		// no hlv on local doc, mark as missing but send current rev as known rev (will be handled as legacy
+		// rev document on changes response handler)
+		base.DebugfCtx(ctx, base.KeyChanges, "Doc %s has no HLV, marking change version %s as missing", base.UD(docid), base.UD(rev))
+		missing = append(missing, rev)
+		possible = append(possible, doc.CurrentRev)
+		return
+	}
 	// parse in coming version, if it's not known to local doc hlv then it is marked as missing, if it is and is a newer version
 	// then it is also marked as missing
 	cvValue, err := ParseVersion(rev)
@@ -3160,6 +3191,13 @@ func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, do
 		// incoming version is dominated by local doc hlv, so it is not missing
 		return
 	}
+
+	// return the local current rev as known rev, this will mean if you have rev 1,2,3 and remote has rev 1,2,3,4,5 then
+	// remote should only send rev 4,5 in rev tree property on the subsequent rev message for this document, we also need to
+	// send cv as first element for delta sync purposes
+	possible = append(possible, doc.HLV.GetCurrentVersionString())
+	possible = append(possible, doc.CurrentRev)
+
 	missing = append(missing, rev)
 	return
 }
@@ -3328,6 +3366,47 @@ func (db *DatabaseCollectionWithUser) CheckProposedVersion(ctx context.Context,
 	}
 }
 
+func (doc *Document) alignRevTreeHistory(ctx context.Context, newDoc *Document, revTreeHistory []string) error {
+	currentRevIndex := len(revTreeHistory)
+	parent := ""
+	for i, revid := range revTreeHistory {
+		if doc.History.contains(revid) {
+			currentRevIndex = i
+			parent = revid
+			break
+		}
+	}
+
+	if parent != doc.CurrentRev {
+		base.DebugfCtx(ctx, base.KeyCRUD, "incoming rev tree history has different history than local doc %s, aligning local doc history with incoming rev tree history", base.UD(doc.ID))
+		// clean local history and make way for incoming history to replace it
+		doc.History = make(RevTree)
+	}
+
+	err := doc.addNewerRevisionsToHistory(newDoc, currentRevIndex, parent, revTreeHistory)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (doc *Document) addNewerRevisionsToHistory(newDoc *Document, currentRevIndex int, parent string, docHistory []string) error {
+	for i := currentRevIndex - 1; i >= 0; i-- {
+		err := doc.History.addRevision(newDoc.ID,
+			RevInfo{
+				ID:      docHistory[i],
+				Parent:  parent,
+				Deleted: i == 0 && newDoc.Deleted})
+
+		if err != nil {
+			return err
+		}
+		parent = docHistory[i]
+	}
+	doc.CurrentRev = parent
+	return nil
+}
+
 const (
 	xattrMacroCas               = "cas"          // SyncData.Cas
 	xattrMacroValueCrc32c       = "value_crc32c" // SyncData.Crc32c

db/crud_test.go

@@ -1773,7 +1773,7 @@ func TestPutExistingCurrentVersion(t *testing.T) {
 		PreviousVersions: pv,
 	}
 
-	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil)
+	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil, false)
 	assertHTTPError(t, err, 409)
 	require.Nil(t, doc)
 	require.Nil(t, cv)
@@ -1783,7 +1783,7 @@ func TestPutExistingCurrentVersion(t *testing.T) {
 	// TODO: because currentRev isn't being updated, storeOldBodyInRevTreeAndUpdateCurrent isn't
 	//  updating the document body.   Need to review whether it makes sense to keep using
 	// storeOldBodyInRevTreeAndUpdateCurrent, or if this needs a larger overhaul to support VV
-	doc, cv, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil)
+	doc, cv, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil, false)
 	require.NoError(t, err)
 	assert.Equal(t, "test", cv.SourceID)
 	assert.Equal(t, incomingVersion, cv.Value)
@@ -1805,7 +1805,7 @@ func TestPutExistingCurrentVersion(t *testing.T) {
 
 	// Attempt to push the same client update, validate server rejects as an already known version and cancels the update.
 	// This case doesn't return error, verify that SyncData hasn't been changed.
-	_, _, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil)
+	_, _, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil, false)
 	require.NoError(t, err)
 	syncData2, err := collection.GetDocSyncData(ctx, "doc1")
 	require.NoError(t, err)
@@ -1851,7 +1851,7 @@ func TestPutExistingCurrentVersionWithConflict(t *testing.T) {
 	}
 
 	// assert that a conflict is correctly identified and the doc and cv are nil
-	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil)
+	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, nil, nil, false)
 	assertHTTPError(t, err, 409)
 	require.Nil(t, doc)
 	require.Nil(t, cv)
@@ -1891,7 +1891,7 @@ func TestPutExistingCurrentVersionWithNoExistingDoc(t *testing.T) {
 		PreviousVersions: pv,
 	}
 	// call PutExistingCurrentVersion with empty existing doc
-	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, &sgbucket.BucketDocument{}, nil)
+	doc, cv, _, err := collection.PutExistingCurrentVersion(ctx, newDoc, incomingHLV, &sgbucket.BucketDocument{}, nil, false)
 	require.NoError(t, err)
 	assert.NotNil(t, doc)
 	// assert on returned CV value

db/import_test.go

@@ -166,7 +166,7 @@ func TestOnDemandImportMou(t *testing.T) {
 				case "PutExistingCurrentVersion":
 					hlv := NewHybridLogicalVector()
 					var legacyRevList []string
-					_, _, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, hlv, rawBucketDoc, legacyRevList)
+					_, _, _, err = collection.PutExistingCurrentVersion(ctx, newDoc, hlv, rawBucketDoc, legacyRevList, false)
 					assertHTTPError(t, err, 409)
 				default:
 					require.FailNow(t, fmt.Sprintf("unexpected funcName: %s", funcName))

db/util_testing.go

@@ -882,7 +882,7 @@ func (c *DatabaseCollectionWithUser) UpsertTestDocWithVersion(ctx context.Contex
 		}
 	}
 
-	doc, _, _, err := c.PutExistingCurrentVersion(ctx, newDoc, newDocHLV, currentBucketDoc, nil)
+	doc, _, _, err := c.PutExistingCurrentVersion(ctx, newDoc, newDocHLV, currentBucketDoc, nil, false)
 	require.NoError(t, err)
 	return doc
 }

db/utilities_hlv_testing.go

@@ -36,7 +36,6 @@ func (v DocVersion) GoString() string {
 	return fmt.Sprintf("DocVersion{RevTreeID:%s,CV:%#v}", v.RevTreeID, v.CV)
 }
 
-// Can maybe be changed to check both rev and cv when CBG-4790, CBG-4791 are implemented
 func (v DocVersion) DocVersionRevTreeEqual(o DocVersion) bool {
 	if v.RevTreeID != o.RevTreeID {
 		return false