CBG-4841 write legacy rev versions with special encoding (#7732)

Author: torcolvin

db/blip_handler.go

@@ -1347,10 +1347,25 @@ func (bh *blipHandler) processRev(rq *blip.Message, stats *processRevStats) (err
 	forceAllowConflictingTombstone := newDoc.Deleted && (!bh.conflictResolver.IsEmpty() || bh.clientType == BLIPClientTypeSGR2)
 	if bh.useHLV() && changeIsVector {
 		_, _, _, err = bh.collection.PutExistingCurrentVersion(bh.loggingCtx, newDoc, incomingHLV, rawBucketDoc, legacyRevList, isBlipRevTreeProperty, bh.conflictResolver)
-	} else if bh.conflictResolver.revTreeConflictResolver != nil {
-		_, _, err = bh.collection.PutExistingRevWithConflictResolution(bh.loggingCtx, newDoc, history, true, bh.conflictResolver.revTreeConflictResolver, forceAllowConflictingTombstone, rawBucketDoc, ExistingVersionWithUpdateToHLV)
 	} else {
-		_, _, err = bh.collection.PutExistingRev(bh.loggingCtx, newDoc, history, revNoConflicts, forceAllowConflictingTombstone, rawBucketDoc, ExistingVersionWithUpdateToHLV)
+		docUpdateEvent := ExistingVersionWithUpdateToHLV
+		if bh.useHLV() {
+			docUpdateEvent = ExistingVersionLegacyRev
+		}
+		opts := putDocOptions{
+			newDoc:                         newDoc,
+			revTreeHistory:                 history,
+			forceAllowConflictingTombstone: forceAllowConflictingTombstone,
+			existingDoc:                    rawBucketDoc,
+			docUpdateEvent:                 docUpdateEvent,
+		}
+		if bh.conflictResolver.revTreeConflictResolver != nil {
+			opts.conflictResolver = bh.conflictResolver.revTreeConflictResolver
+			opts.noConflicts = true
+		} else {
+			opts.noConflicts = revNoConflicts
+		}
+		_, _, err = bh.collection.PutExistingRevWithConflictResolution(bh.loggingCtx, opts)
 	}
 	if err != nil {
 		return err

db/crud.go

@@ -1016,9 +1016,22 @@ func (db *DatabaseCollectionWithUser) updateHLV(ctx context.Context, d *Document
 		}
 		// update the cvCAS on the SGWrite event too
 		d.HLV.CurrentVersionCAS = expandMacroCASValueUint64
+	case ExistingVersionLegacyRev:
+		revTreeEncodedCV, err := LegacyRevToRevTreeEncodedVersion(d.GetRevTreeID())
+		if err != nil {
+			return nil, err
+		}
+		err = d.HLV.AddVersion(revTreeEncodedCV)
+		if err != nil {
+			return nil, err
+		}
+		// update the cvCAS on the SGWrite event too
+		d.HLV.CurrentVersionCAS = expandMacroCASValueUint64
 	case NoHLVUpdateForTest:
 		// no hlv update event for testing purposes only (used to simulate pre upgraded write)
 		return d, nil
+	default:
+		return nil, base.RedactErrorf("Unexpected docUpdateEvent %v in updateHLV for doc %s", docUpdateEvent, base.UD(d.ID))
 	}
 	d.SyncData.SetCV(d.HLV)
 	return d, nil
@@ -1449,24 +1462,45 @@ func (db *DatabaseCollectionWithUser) PutExistingCurrentVersion(ctx context.Cont
 
 // Adds an existing revision to a document along with its history (list of rev IDs.)
 func (db *DatabaseCollectionWithUser) PutExistingRev(ctx context.Context, newDoc *Document, docHistory []string, noConflicts bool, forceAllConflicts bool, existingDoc *sgbucket.BucketDocument, docUpdateEvent DocUpdateType) (doc *Document, newRevID string, err error) {
-	return db.PutExistingRevWithConflictResolution(ctx, newDoc, docHistory, noConflicts, nil, forceAllConflicts, existingDoc, docUpdateEvent)
+	opts := putDocOptions{
+		newDoc:                         newDoc,
+		revTreeHistory:                 docHistory,
+		noConflicts:                    noConflicts,
+		forceAllowConflictingTombstone: forceAllConflicts,
+		existingDoc:                    existingDoc,
+		docUpdateEvent:                 docUpdateEvent,
+	}
+	return db.PutExistingRevWithConflictResolution(ctx, opts)
+}
+
+// putDocOptions encapsulates the options for putting a document revision.
+type putDocOptions struct {
+	newDoc                         *Document                // the next contents of the incoming document
+	revTreeHistory                 []string                 // list of rev tree IDs. The first entry must be the revtree ID that will be added.
+	docUpdateEvent                 DocUpdateType            // new write, existing write, import etc
+	noConflicts                    bool                     // If true, return 409 on any conflict writes
+	forceAllowConflictingTombstone bool                     // If true, do not flag an incoming tombstone as a conflict if the existing document is a tombstone
+	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
 }
 
-// PutExistingRevWithConflictResolution Adds an existing revision to a document along with its history (list of rev IDs.)
+// PutExistingRevWithConflictResolution adds an existing revision to a document along with its history.
 // If this new revision would result in a conflict:
 //  1. If noConflicts == false, the revision will be added to the rev tree as a conflict
 //  2. If noConflicts == true and a conflictResolverFunc is not provided, a 409 conflict error will be returned
 //  3. If noConflicts == true and a conflictResolverFunc is provided, conflicts will be resolved and the result added to the document.
-func (db *DatabaseCollectionWithUser) PutExistingRevWithConflictResolution(ctx context.Context, newDoc *Document, docHistory []string, noConflicts bool, conflictResolver *ConflictResolver, forceAllowConflictingTombstone bool, existingDoc *sgbucket.BucketDocument, docUpdateEvent DocUpdateType) (doc *Document, newRevID string, err error) {
-	newRev := docHistory[0]
+func (db *DatabaseCollectionWithUser) PutExistingRevWithConflictResolution(ctx context.Context, opts putDocOptions) (doc *Document, newRevID string, err error) {
+	newRev := opts.revTreeHistory[0]
 	generation, _ := ParseRevID(ctx, newRev)
 	if generation < 0 {
 		return nil, "", base.HTTPErrorf(http.StatusBadRequest, "Invalid revision ID")
 	}
 
+	newDoc := opts.newDoc
+	docHistory := opts.revTreeHistory
 	allowImport := db.UseXattrs()
 	updateRevCache := true
-	doc, _, err = db.updateAndReturnDoc(ctx, newDoc.ID, allowImport, &newDoc.DocExpiry, nil, docUpdateEvent, existingDoc, false, updateRevCache, func(doc *Document) (resultDoc *Document, resultAttachmentData updatedAttachments, createNewRevIDSkipped bool, updatedExpiry *uint32, resultErr error) {
+	doc, _, err = db.updateAndReturnDoc(ctx, newDoc.ID, allowImport, &newDoc.DocExpiry, nil, opts.docUpdateEvent, opts.existingDoc, false, updateRevCache, func(doc *Document) (resultDoc *Document, resultAttachmentData updatedAttachments, createNewRevIDSkipped bool, updatedExpiry *uint32, resultErr error) {
 		// (Be careful: this block can be invoked multiple times if there are races!)
 
 		var isSgWrite bool
@@ -1507,13 +1541,13 @@ func (db *DatabaseCollectionWithUser) PutExistingRevWithConflictResolution(ctx c
 		// Conflict-free mode check
 
 		// We only bypass conflict resolution for incoming tombstones if the local doc is also a tombstone
-		allowConflictingTombstone := forceAllowConflictingTombstone && doc.IsDeleted()
+		allowConflictingTombstone := opts.forceAllowConflictingTombstone && doc.IsDeleted()
 
-		if !allowConflictingTombstone && db.IsIllegalConflict(ctx, doc, parent, newDoc.Deleted, noConflicts, docHistory) {
-			if conflictResolver == nil {
+		if !allowConflictingTombstone && db.IsIllegalConflict(ctx, doc, parent, newDoc.Deleted, opts.noConflicts, docHistory) {
+			if opts.conflictResolver == nil {
 				return nil, nil, false, nil, base.HTTPErrorf(http.StatusConflict, "Document revision conflict")
 			}
-			_, updatedHistory, err := db.resolveConflict(ctx, doc, newDoc, docHistory, conflictResolver)
+			_, updatedHistory, err := db.resolveConflict(ctx, doc, newDoc, docHistory, opts.conflictResolver)
 			if err != nil {
 				base.InfofCtx(ctx, base.KeyCRUD, "Error resolving conflict for %s: %v", base.UD(doc.ID), err)
 				return nil, nil, false, nil, err

db/database.go

@@ -53,6 +53,7 @@ const (
 	Import DocUpdateType = iota
 	NewVersion
 	ExistingVersion
+	ExistingVersionLegacyRev
 	ExistingVersionWithUpdateToHLV
 	NoHLVUpdateForTest
 )

db/hybrid_logical_vector.go

@@ -15,6 +15,7 @@ import (
 	"fmt"
 	"iter"
 	"maps"
+	"math/bits"
 	"slices"
 	"sort"
 	"strconv"
@@ -24,6 +25,10 @@ import (
 	"github.com/couchbase/sync_gateway/base"
 )
 
+// encodedRevTreeSourceID is a base64 encoded value representing a revision that came from a 4.x client with a
+// revtree ID.
+const encodedRevTreeSourceID = "Revision+Tree+Encoding"
+
 type HLVVersions map[string]uint64 // map of source ID to version uint64 version value
 
 // sorted will iterate through the map returning entries in a stable sorted order. Used by testing to make it easier
@@ -518,7 +523,7 @@ func extractHLVFromBlipString(versionVectorStr string) (*HybridLogicalVector, []
 		return nil, nil, err
 	}
 	if legacyRevs != nil {
-		return nil, nil, fmt.Errorf("invalid hlv in changes message received, legacy revID found in cv: %q", vectorFields[0])
+		return nil, nil, fmt.Errorf("invalid hlv in changes message received, legacy revIDs found in cv %s", versionVectorStr)
 	}
 	for i, v := range cvmvList {
 		switch i {
@@ -626,7 +631,7 @@ func isLegacyRev(rev string) bool {
 	return true
 }
 
-// Helper functions for version source and value encoding
+// EncodeSource encodes a source ID in base64 for storage.
 func EncodeSource(source string) string {
 	return base64.StdEncoding.EncodeToString([]byte(source))
 }
@@ -901,3 +906,27 @@ func remoteWinsConflictResolutionForHLV(ctx context.Context, docID string, local
 	base.DebugfCtx(ctx, base.KeyVV, "resolved conflict for doc %s in favour of remote wins, resulting HLV: %v", base.UD(docID), newHLV)
 	return newHLV, nil
 }
+
+// LegacyRevToRevTreeEncodedVersion creates a version that has a specific source ID that can be recognized. The version is made up of:
+//
+// - The upper 24 bits of the version are the generation.
+// - The lower 40 bits of the version are the first 40 bits of the digest, which is right padded.
+func LegacyRevToRevTreeEncodedVersion(legacyRev string) (Version, error) {
+	generation, digest, err := parseRevID(legacyRev)
+	if err != nil {
+		return Version{}, err
+	}
+	// trim to 40 bits (10 hex characters)
+	if len(digest) > 10 {
+		digest = digest[:10]
+	}
+	value, err := strconv.ParseUint(digest, 16, 64)
+	if err != nil {
+		return Version{}, err
+	}
+	value = value << (40 - bits.Len64(value)) // right pad zeros
+	return Version{
+		SourceID: encodedRevTreeSourceID,
+		Value:    (uint64(generation) << 40) | value,
+	}, nil
+}

db/hybrid_logical_vector_test.go

@@ -1660,3 +1660,38 @@ func TestLocalWinsConflictResolutionForHLV(t *testing.T) {
 		})
 	}
 }
+
+func TestLegacyRevToVersion(t *testing.T) {
+	testCases := []struct {
+		legacyRev string
+		version   string
+	}{
+		{
+			legacyRev: "1-abcd",
+			version:   "1abcd000000@Revision+Tree+Encoding",
+		},
+		{
+			legacyRev: "12345-e0c8012361e94df6a1e1c2977169480e",
+			version:   "3039e0c8012361@Revision+Tree+Encoding",
+		},
+		{
+			legacyRev: "16777215-abcd",
+			version:   "ffffffabcd000000@Revision+Tree+Encoding",
+		},
+		{
+			legacyRev: "16777216-abcd",
+			version:   "abcd000000@Revision+Tree+Encoding",
+		},
+		{
+			legacyRev: "16777217-abcd",
+			version:   "1abcd000000@Revision+Tree+Encoding",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.legacyRev, func(t *testing.T) {
+			version, err := LegacyRevToRevTreeEncodedVersion(tc.legacyRev)
+			require.NoError(t, err)
+			assert.Equal(t, tc.version, version.String())
+		})
+	}
+}

db/revision.go

@@ -365,28 +365,38 @@ func genOfRevID(ctx context.Context, revid string) int {
 	return generation
 }
 
-// Splits a revision ID into generation number and hex digest.
-func ParseRevID(ctx context.Context, revid string) (int, string) {
-	if revid == "" {
+// ParseRevID splits a revision ID into generation number and hex digest. Logs a warning and returns -1, "" if the revID is invalid.
+func ParseRevID(ctx context.Context, revID string) (int, string) {
+	if revID == "" {
 		return 0, ""
 	}
+	generation, digest, err := parseRevID(revID)
+	if err != nil {
+		base.WarnfCtx(ctx, "%s", err)
+		return -1, ""
+	}
+	return generation, digest
+}
+
+// parseRevID splits a revision ID into generation number and hex digest. Returns an error if the revID is invalid.
+func parseRevID(revid string) (int, string, error) {
+	if revid == "" {
+		return 0, "", errors.New("empty revID passed to parseRevID")
+	}
 
 	idx := strings.Index(revid, "-")
 	if idx == -1 {
-		base.WarnfCtx(ctx, "parseRevID found no separator in rev %q", revid)
-		return -1, ""
+		return -1, "", fmt.Errorf("parseRevID found no separator in rev %q", revid)
 	}
 
 	gen, err := strconv.Atoi(revid[:idx])
 	if err != nil {
-		base.WarnfCtx(ctx, "parseRevID unexpected generation in rev %q: %s", revid, err)
-		return -1, ""
+		return -1, "", fmt.Errorf("parseRevID unexpected generation in rev %q: %s", revid, err)
 	} else if gen < 1 {
-		base.WarnfCtx(ctx, "parseRevID unexpected generation in rev %q", revid)
-		return -1, ""
+		return -1, "", fmt.Errorf("parseRevID unexpected generation in rev %q", revid)
 	}
 
-	return gen, revid[idx+1:]
+	return gen, revid[idx+1:], nil
 }
 
 // compareRevIDs compares the two rev IDs and returns: