[4.0.0 backport] CBG-4886: delta sync for legacy rev mutations over ISGR will not send deltas (#7806)

Author: gregns1

db/blip_handler.go

@@ -1112,13 +1112,11 @@ 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 {
 			legacyRevList = append(legacyRevList, strings.Split(revTree, ",")...)
-			isBlipRevTreeProperty = true
 			if len(legacyRevList) > 0 {
 				newDoc.RevID = legacyRevList[0]
 			}
@@ -1352,7 +1350,7 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
 			ExistingDoc:                    rawBucketDoc,
 			NewDocHLV:                      incomingHLV,
 			ConflictResolver:               bh.conflictResolver.hlvConflictResolver,
-			AlignRevTrees:                  isBlipRevTreeProperty,
+			ISGRWrite:                      bh.clientType == BLIPClientTypeSGR2,
 		}
 		_, _, _, err = bh.collection.PutExistingCurrentVersion(bh.loggingCtx, opts)
 	} else {

db/blip_sync_context.go

@@ -597,11 +597,12 @@ func (bsc *BlipSyncContext) sendDelta(ctx context.Context, sender *blip.Sender,
 	properties := blipRevMessageProperties(history, revDelta.ToDeleted, seq, "", revTreeProperty)
 	properties[RevMessageDeltaSrc] = deltaSrcRevID
 
-	base.DebugfCtx(ctx, base.KeySync, "Sending rev %q %s as delta. DeltaSrc:%s", base.UD(docID), revDelta.ToRevID, deltaSrcRevID)
 	if bsc.useHLV() {
+		base.DebugfCtx(ctx, base.KeySync, "Sending rev %q %s as delta. DeltaSrc:%s", base.UD(docID), revDelta.ToCV, deltaSrcRevID)
 		return bsc.sendRevisionWithProperties(ctx, sender, docID, revDelta.ToCV, collectionIdx, revDelta.DeltaBytes, revDelta.AttachmentStorageMeta,
 			properties, seq, resendFullRevisionFunc)
 	} else {
+		base.DebugfCtx(ctx, base.KeySync, "Sending rev %q %s as delta. DeltaSrc:%s", base.UD(docID), revDelta.ToRevID, deltaSrcRevID)
 		return bsc.sendRevisionWithProperties(ctx, sender, docID, revDelta.ToRevID, collectionIdx, revDelta.DeltaBytes, revDelta.AttachmentStorageMeta,
 			properties, seq, resendFullRevisionFunc)
 	}
@@ -872,7 +873,7 @@ func (bsc *BlipSyncContext) getKnownRevs(ctx context.Context, docID string, know
 	// revtree clients. For HLV clients, use the cv as deltaSrc
 	if bsc.useDeltas && len(knownRevsArray) > 0 {
 		if revID, ok := knownRevsArray[0].(string); ok {
-			if bsc.useHLV() {
+			if bsc.useHLV() && !base.IsRevTreeID(revID) {
 				// extract cv from the known revs array
 				msgHLV, _, deltaSrcErr := extractHLVFromBlipString(revID)
 				if deltaSrcErr != nil {

db/crud.go

@@ -1282,7 +1282,7 @@ func (db *DatabaseCollectionWithUser) Put(ctx context.Context, docid string, bod
 //   - NewDocHLV: new incoming doc's HLV
 //   - ExistingDoc: existing doc in bucket (if present)
 //   - RevTreeHistory: list of revID's from the incoming docs history (including docs current rev).
-//   - AlignRevTrees: if this is true then we will align the new write with the incoming docs rev tree. If this is
+//   - ISGRWrite: if this is true then we will align the new write with the incoming docs rev tree. If this is
 //     false and len(RevTreeHistory) > 0 then this means the local version of this doc does not have an HLV so this parameter
 //     will be used to check for conflicts.
 func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Context, opts PutDocOptions) (doc *Document, cv *Version, newRevID string, err error) {
@@ -1339,7 +1339,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 			// 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 !opts.AlignRevTrees {
+			if !opts.ISGRWrite {
 				newGeneration = prevGeneration + 1
 			} else {
 				newGeneration = prevGeneration
@@ -1373,7 +1373,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 				doc.HLV = NewHybridLogicalVector()
 			}
 			doc.HLV.UpdateWithIncomingHLV(opts.NewDocHLV)
-			if opts.AlignRevTrees {
+			if opts.ISGRWrite {
 				err := doc.alignRevTreeHistoryForHLVWrite(ctx, db, opts.NewDoc, opts.RevTreeHistory, opts.ForceAllowConflictingTombstone)
 				if err != nil {
 					return nil, nil, false, nil, err
@@ -1393,7 +1393,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 				// update hlv for all newer incoming source version pairs
 				doc.HLV.UpdateWithIncomingHLV(opts.NewDocHLV)
 				// the new document has a dominating hlv, so we can just update local revtree with incoming revtree
-				if !opts.AlignRevTrees {
+				if !opts.ISGRWrite {
 					previousRevTreeID = doc.GetRevTreeID()
 				} else {
 					// align rev tree here for ISGR replications
@@ -1404,7 +1404,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 				}
 			case HLVConflict:
 				// if we have been supplied a rev tree from cbl, perform conflict check on rev tree history
-				if len(opts.RevTreeHistory) > 0 && !opts.AlignRevTrees {
+				if len(opts.RevTreeHistory) > 0 && !opts.ISGRWrite {
 					parent, currentRevIndex, err := db.revTreeConflictCheck(ctx, opts.RevTreeHistory, doc, opts.NewDoc.Deleted)
 					if err != nil {
 						base.DebugfCtx(ctx, base.KeyCRUD, "conflict detected between the two HLV's for doc %s, and conflict found in rev tree history", base.UD(doc.ID))
@@ -1448,7 +1448,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 		// if we have revtree history available and we are communicating with CBL, we must update the rev tree
 		// to include the new to us revisions from the incoming rev tree history skipping history check given conflict
 		// check is done at this point.
-		if len(opts.RevTreeHistory) > 0 && !opts.AlignRevTrees && !revTreeAlignedForCBL {
+		if len(opts.RevTreeHistory) > 0 && !opts.ISGRWrite && !revTreeAlignedForCBL {
 			addNewRevErr := doc.alignRevTreeHistoryForHLVWrite(ctx, db, opts.NewDoc, opts.RevTreeHistory, true)
 			if addNewRevErr != nil {
 				return nil, nil, false, nil, addNewRevErr
@@ -1463,7 +1463,7 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 
 		// generate rev id for new arriving doc
 		var newRev string
-		if !opts.AlignRevTrees {
+		if !opts.ISGRWrite {
 			// create a new revID for incoming write
 			strippedBody, _ := StripInternalProperties(opts.NewDoc._body)
 			encoding, err := base.JSONMarshalCanonical(strippedBody)
@@ -1521,7 +1521,7 @@ type PutDocOptions struct {
 	ConflictResolver               *ConflictResolver        // If provided, will be used to resolve conflicts if NoConflicts is false and a conflict is detected
 	ExistingDoc                    *sgbucket.BucketDocument // optional, prevents fetching the document from the bucket
 	NewDocHLV                      *HybridLogicalVector     // incoming doc HLV if known
-	AlignRevTrees                  bool                     // if true, the rev tree history is from ISGR only revTree replication message property (> v4 protocol only) and will use this to align rev trees
+	ISGRWrite                      bool                     // if true, the write is from ISGR and will use rev tree history to align rev trees
 }
 
 // PutExistingRevWithConflictResolution adds an existing revision to a document along with its history.
@@ -3455,38 +3455,42 @@ func (c *DatabaseCollection) checkForUpgrade(ctx context.Context, key string, un
 func legacyRevToHybridLogicalVector(docID, revID string) (hlv *HybridLogicalVector, err error) {
 	version, err := LegacyRevToRevTreeEncodedVersion(revID)
 	if err != nil {
-		return nil, fmt.Errorf("error parsing legacy revID %q to version for doc %s: %v", revID, base.UD(docID), err)
+		return nil, base.RedactErrorf("error parsing legacy revID %q to version for doc %s: %v", revID, base.UD(docID), err)
 	}
 	hlv = NewHybridLogicalVector()
 	err = hlv.AddVersion(version)
 	if err != nil {
-		return nil, fmt.Errorf("error adding version to hlv for doc %s: %v", base.UD(docID), err)
+		return nil, base.RedactErrorf("error adding version to hlv for doc %s: %v", base.UD(docID), err)
 	}
 	return hlv, nil
 }
 
 // parseIncomingChange will parse incoming change version. If the change is legacy rev it will convert the revID to a CV
-// otherwise it will parse the version string.
-func parseIncomingChange(docid, rev string) (cvValue Version, err error) {
+// otherwise it will parse the version string and will mark change as legacy rev.
+func parseIncomingChange(docid, rev string) (cvValue Version, legacyRev bool, err error) {
 	if base.IsRevTreeID(rev) {
 		cvValue, err = LegacyRevToRevTreeEncodedVersion(rev)
 		if err != nil {
-			return Version{}, base.RedactErrorf("error parsing legacy revID %q to version for doc %s: %v", base.UD(rev), base.UD(docid), err)
+			return Version{}, legacyRev, base.RedactErrorf("error parsing legacy revID %q to version for doc %s: %v", base.UD(rev), base.UD(docid), err)
 		}
+		legacyRev = true
 	} else {
 		cvValue, err = ParseVersion(rev)
 		if err != nil {
-			return Version{}, base.RedactErrorf("error parsing change version for doc %s: %v", base.UD(docid), err)
+			return Version{}, legacyRev, base.RedactErrorf("error parsing change version for doc %s: %v", base.UD(docid), err)
 		}
 	}
-	return cvValue, nil
+	return cvValue, legacyRev, nil
 }
 
 func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, docid, rev string) (missing, possible []string) {
 	if strings.HasPrefix(docid, "_design/") && db.user != nil {
 		return // Users can't upload design docs, so ignore them
 	}
 
+	localIsLegacy := false
+	changeIsLegacy := false
+
 	syncData, hlv, err := db.GetDocSyncDataNoImport(ctx, docid, DocUnmarshalSync)
 	if err != nil {
 		if !base.IsDocNotFoundError(err) && !base.IsXattrNotFoundError(err) {
@@ -3503,10 +3507,15 @@ func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, do
 			return
 		}
 	}
+	if hlv.HasRevEncodedCV() {
+		// if local hlv has revID encoded CV given to it mark it as legacy change for the purposes of sending back known revs
+		// in legacy format. This will ensure delta sync works correctly replicating legacy revs
+		localIsLegacy = true
+	}
 	// 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
 	var cvValue Version
-	cvValue, err = parseIncomingChange(docid, rev)
+	cvValue, changeIsLegacy, err = parseIncomingChange(docid, rev)
 	if err != nil {
 		base.WarnfCtx(ctx, "%s", err)
 		missing = append(missing, rev)
@@ -3520,8 +3529,11 @@ func (db *DatabaseCollectionWithUser) CheckChangeVersion(ctx context.Context, do
 
 	// 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, hlv.GetCurrentVersionString())
+	// send cv as first element for delta sync purposes. Only send CV if this is not legacy rev change
+	if !localIsLegacy && !changeIsLegacy {
+		// we should only send CV in response when we are communicating with HLV's both sides of the replication
+		possible = append(possible, hlv.GetCurrentVersionString())
+	}
 	possible = append(possible, syncData.GetRevTreeID())
 
 	missing = append(missing, rev)

db/revision_cache_lru.go

@@ -727,6 +727,13 @@ func (value *revCacheValue) load(ctx context.Context, backingStore RevisionCache
 			if hlv != nil {
 				value.cv = *hlv.ExtractCurrentVersionFromHLV()
 				value.hlvHistory = hlv.ToHistoryForHLV()
+			} else if value.err == nil {
+				// if hlv is nil its a legacy rev, we need to create a CV from the revID
+				encodedCV, err := LegacyRevToRevTreeEncodedVersion(value.revID)
+				if err != nil {
+					return docRev, false, err
+				}
+				value.cv = encodedCV
 			}
 		}
 	}

rest/replicatortest/replicator_test_legacy_rev_test.go

@@ -1094,3 +1094,184 @@ func TestActiveReplicatorConflictPreUpgradedVersionOneSide(t *testing.T) {
 		})
 	}
 }
+
+func TestActiveReplicatorDeltaSyncWhenBothSidesLegacy(t *testing.T) {
+	base.RequireNumTestBuckets(t, 2)
+	base.SetUpTestLogging(t, base.LevelDebug, base.KeyAll)
+
+	if !base.IsEnterpriseEdition() {
+		t.Skip("Delta sync only supported in EE")
+	}
+
+	const username = "alice"
+
+	// Passive (SGW2 in diagram above)
+	rt2 := rest.NewRestTester(t,
+		&rest.RestTesterConfig{
+			SyncFn: channels.DocChannelsSyncFunction,
+			DatabaseConfig: &rest.DatabaseConfig{DbConfig: rest.DbConfig{
+				Name: "passivedb",
+				DeltaSync: &rest.DeltaSyncConfig{
+					Enabled: base.Ptr(true),
+				},
+			}},
+		})
+	defer rt2.Close()
+
+	rt2.CreateUser(username, []string{username})
+
+	// Active (SGW1 in diagram above)
+	rt1 := rest.NewRestTester(t,
+		&rest.RestTesterConfig{
+			SyncFn: channels.DocChannelsSyncFunction,
+			DatabaseConfig: &rest.DatabaseConfig{DbConfig: rest.DbConfig{
+				Name: "activedb",
+				DeltaSync: &rest.DeltaSyncConfig{
+					Enabled: base.Ptr(true),
+				},
+			}},
+		})
+	defer rt1.Close()
+	ctx1 := rt1.Context()
+
+	docIDToPush := rest.SafeDocumentName(t, t.Name()+"_push")
+
+	// create doc on rt1 with one revision
+	bodyRT1 := db.Body{"channels": []string{username}, "source": "rt1"}
+	rt1InitDoc := rt1.CreateDocNoHLV(docIDToPush, bodyRT1)
+	legacyInitRevRt1 := rt1InitDoc.GetRevTreeID()
+	// create another rev to ensure we have a rev to delta from
+	bodyRT1 = db.Body{db.BodyRev: legacyInitRevRt1, "channels": []string{username}, "source": "rt1"}
+	rt1InitDoc = rt1.CreateDocNoHLV(docIDToPush, bodyRT1)
+	legacyRevRt1 := rt1InitDoc.GetRevTreeID()
+
+	// create rev on rt2 that will resolve to same revID as rev one above simulating the following:
+	// 1. doc created on rt1, pushed to rt2
+	// 2. doc updated on rt1 to create rev2, but upgrade happens before being pushed to rt2
+	// 3. doc is pushed post upgrade to rt2 and the delta from rev1 to rev2 is sent
+	bodyRT2 := db.Body{"channels": []string{username}, "source": "rt1"}
+	rt2InitDoc := rt2.CreateDocNoHLV(docIDToPush, bodyRT2)
+	legacyRevRt2 := rt2InitDoc.GetRevTreeID()
+
+	require.Equal(t, legacyInitRevRt1, legacyRevRt2)
+
+	stats, err := base.SyncGatewayStats.NewDBStats(t.Name(), false, false, false, nil, nil)
+	require.NoError(t, err)
+	replicationStats, err := stats.DBReplicatorStats(t.Name())
+	require.NoError(t, err)
+
+	ar, err := db.NewActiveReplicator(ctx1, &db.ActiveReplicatorConfig{
+		ID:          t.Name(),
+		Direction:   db.ActiveReplicatorTypePush,
+		RemoteDBURL: userDBURL(rt2, username),
+		ActiveDB: &db.Database{
+			DatabaseContext: rt1.GetDatabase(),
+		},
+		ChangesBatchSize:    200,
+		Continuous:          true,
+		ReplicationStatsMap: replicationStats,
+		CollectionsEnabled:  !rt1.GetDatabase().OnlyDefaultCollection(),
+		DeltasEnabled:       true,
+	})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, ar.Stop())
+	}()
+
+	// Start the replicator
+	require.NoError(t, ar.Start(ctx1))
+
+	rt2.WaitForLegacyRev(docIDToPush, legacyRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`))
+
+	base.RequireWaitForStat(t, func() int64 {
+		return replicationStats.PushDeltaSentCount.Value()
+	}, 1)
+}
+
+func TestDeltaSyncWhenOneSideHasEncodedCV(t *testing.T) {
+	base.RequireNumTestBuckets(t, 2)
+
+	if !base.IsEnterpriseEdition() {
+		t.Skip("Delta sync only supported in EE")
+	}
+
+	const username = "alice"
+
+	// Passive (SGW2 in diagram above)
+	rt2 := rest.NewRestTester(t,
+		&rest.RestTesterConfig{
+			SyncFn: channels.DocChannelsSyncFunction,
+			DatabaseConfig: &rest.DatabaseConfig{DbConfig: rest.DbConfig{
+				Name: "passivedb",
+				DeltaSync: &rest.DeltaSyncConfig{
+					Enabled: base.Ptr(true),
+				},
+			}},
+		})
+	defer rt2.Close()
+
+	rt2.CreateUser(username, []string{username})
+
+	// Active (SGW1 in diagram above)
+	rt1 := rest.NewRestTester(t,
+		&rest.RestTesterConfig{
+			SyncFn: channels.DocChannelsSyncFunction,
+			DatabaseConfig: &rest.DatabaseConfig{DbConfig: rest.DbConfig{
+				Name: "activedb",
+				DeltaSync: &rest.DeltaSyncConfig{
+					Enabled: base.Ptr(true),
+				},
+			}},
+		})
+	defer rt1.Close()
+	ctx1 := rt1.Context()
+
+	docIDToPush := rest.SafeDocumentName(t, t.Name()+"_push")
+
+	// create doc on rt1 with one revision
+	bodyRT1 := db.Body{"channels": []string{username}, "source": "rt1"}
+	rt1InitDoc := rt1.CreateDocNoHLV(docIDToPush, bodyRT1)
+	legacyInitRevRt1 := rt1InitDoc.GetRevTreeID()
+
+	stats, err := base.SyncGatewayStats.NewDBStats(t.Name(), false, false, false, nil, nil)
+	require.NoError(t, err)
+	replicationStats, err := stats.DBReplicatorStats(t.Name())
+	require.NoError(t, err)
+
+	ar, err := db.NewActiveReplicator(ctx1, &db.ActiveReplicatorConfig{
+		ID:          t.Name(),
+		Direction:   db.ActiveReplicatorTypePush,
+		RemoteDBURL: userDBURL(rt2, username),
+		ActiveDB: &db.Database{
+			DatabaseContext: rt1.GetDatabase(),
+		},
+		ChangesBatchSize:    200,
+		Continuous:          true,
+		ReplicationStatsMap: replicationStats,
+		CollectionsEnabled:  !rt1.GetDatabase().OnlyDefaultCollection(),
+		DeltasEnabled:       true,
+	})
+	require.NoError(t, err)
+	defer func() {
+		require.NoError(t, ar.Stop())
+	}()
+
+	// Start the replicator
+	require.NoError(t, ar.Start(ctx1))
+
+	rt2.WaitForLegacyRev(docIDToPush, legacyInitRevRt1, []byte(`{"source":"rt1","channels":["alice"]}`))
+
+	// flush revision cache to remove old reference to rev 1 in rev cache
+	rt1.GetDatabase().FlushRevisionCacheForTest()
+
+	// update doc on rt1 to create a second revision with HLV
+	// This should:
+	// 1. update doc on rt1 to give HLV based of rt1 sourceID
+	// 2. push doc to rt2 with delta from rev1 to rev2
+	upgradeVersion := rt1.UpdateDoc(docIDToPush, db.DocVersion{RevTreeID: legacyInitRevRt1}, `{"channels": ["alice"], "source": "rt1-updated"}`)
+	rt1.WaitForVersion(docIDToPush, upgradeVersion)
+
+	base.RequireWaitForStat(t, func() int64 {
+		return replicationStats.PushDeltaSentCount.Value()
+	}, 1)
+}