restore common signatures

Author: torcolvin

db/change_cache.go

@@ -395,18 +395,13 @@ func (c *changeCache) DocChanged(event sgbucket.FeedEvent, docType DocumentType)
 		if syncData == nil {
 			return
 		}
-		var cv *Version
 		vv := doc.Xattrs[base.VvXattrName]
 		if len(vv) > 0 {
-			cv, err = getCurrentVersionFromVVXattr(vv)
-			if err != nil {
-				base.WarnfCtx(ctx, "Error unmarshalling %s xattr (%q) for doc %q: %v", base.VvXattrName, string(vv), base.UD(docID), err)
+			isSGWrite, _, _ := syncData.IsSGWrite(ctx, event.Cas, doc.Body, rawUserXattr, base.Ptr(rawHLV(vv)))
+			if !isSGWrite {
+				return
 			}
 		}
-		isSGWrite, _, _ := syncData.IsSGWrite(event.Cas, doc.Body, rawUserXattr, cv)
-		if !isSGWrite {
-			return
-		}
 	}
 
 	// If not using xattrs and no sync metadata found, check whether we're mid-upgrade and attempting to read a doc w/ metadata stored in xattr

db/crud_test.go

@@ -2191,13 +2191,14 @@ func TestIsSGWrite(t *testing.T) {
 		}
 	})
 	t.Run("no _sync.rev.src", func(t *testing.T) {
+		// this is a corrupt _sync.rev, so assume that it was a _vv at some point and import just in case
 		doc := createNewTestDocument(t, db, body)
 		doc.RevAndVersion.CurrentSource = ""
 		doc.Cas = 1 // force mismatch cas
 		for _, testCase := range testCases {
 			t.Run(testCase.name, func(t *testing.T) {
 				isSGWrite, _, _ := doc.IsSGWrite(ctx, testCase.docBody)
-				require.True(t, isSGWrite, "Expected doc to be identified as SG write for body %q", string(testCase.docBody))
+				require.False(t, isSGWrite, "Expected doc not to be identified as SG write for body %q since _sync.rev.src is empty", string(testCase.docBody))
 			})
 		}
 	})
@@ -2208,7 +2209,7 @@ func TestIsSGWrite(t *testing.T) {
 		for _, testCase := range testCases {
 			t.Run(testCase.name, func(t *testing.T) {
 				isSGWrite, _, _ := doc.IsSGWrite(ctx, testCase.docBody)
-				require.True(t, isSGWrite, "Expected doc to be identified as SG write for body %q since _sync.rev.ver is empty", string(testCase.docBody))
+				require.False(t, isSGWrite, "Expected doc not to be identified as SG write for body %q since _sync.rev.ver is empty", string(testCase.docBody))
 			})
 		}
 	})
@@ -2239,12 +2240,12 @@ func TestSyncDataCVEqual(t *testing.T) {
 			syncData: SyncData{
 				RevAndVersion: channels.RevAndVersion{
 					CurrentSource:  "testSourceID",
-					CurrentVersion: "0x1",
+					CurrentVersion: "0x0100000000000000",
 				},
 			},
 			cvSourceID: "testSourceID",
 			cvValue:    1,
-			cvEqual:    false,
+			cvEqual:    true,
 		},
 		{
 			name: "syncData sourceID mismatch",
@@ -2256,7 +2257,7 @@ func TestSyncDataCVEqual(t *testing.T) {
 			},
 			cvSourceID: "testSourceID2",
 			cvValue:    1,
-			cvEqual:    true,
+			cvEqual:    false,
 		},
 		{
 			name: "syncData sourceID mismatch",
@@ -2268,24 +2269,16 @@ func TestSyncDataCVEqual(t *testing.T) {
 			},
 			cvSourceID: "testSourceID",
 			cvValue:    1,
-			cvEqual:    true,
-		},
-		{
-			name: "syncData equal",
-			syncData: SyncData{
-				RevAndVersion: channels.RevAndVersion{
-					CurrentSource:  "sourceID",
-					CurrentVersion: "0x1",
-				},
-			},
-			cvSourceID: "sourceID",
-			cvValue:    1,
-			cvEqual:    true,
+			cvEqual:    false,
 		},
 	}
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
-			assert.True(t, testCase.syncData.CVEqual(testCase.cvSourceID, testCase.cvValue))
+			cv := Version{
+				SourceID: testCase.cvSourceID,
+				Value:    testCase.cvValue,
+			}
+			require.Equal(t, testCase.cvEqual, testCase.syncData.CVEqual(cv))
 		})
 	}
 }

db/document.go

@@ -662,15 +662,15 @@ func HasUserXattrChanged(userXattr []byte, prevUserXattrHash string) bool {
 }
 
 // CVEqual returns true if the provided CV does not match _sync.rev.ver and _sync.rev.src. The caller is responsible for testing if the values are non-empty.
-func (s *SyncData) CVEqual(cvSourceID string, cvVersion uint64) bool {
-	if cvSourceID != s.RevAndVersion.CurrentSource {
-		return true
+func (s *SyncData) CVEqual(cv Version) bool {
+	if cv.SourceID != s.RevAndVersion.CurrentSource {
+		return false
 	}
-	return cvVersion != base.HexCasToUint64(s.RevAndVersion.CurrentVersion)
+	return cv.Value == base.HexCasToUint64(s.RevAndVersion.CurrentVersion)
 }
 
 // IsSGWrite determines if a document was written by Sync Gateway or via an SDK. CV is an optional parameter to check. This would represent _vv.ver and _vv.src
-func (s *SyncData) IsSGWrite(cas uint64, rawBody []byte, rawUserXattr []byte, cv *Version) (isSGWrite bool, crc32Match bool, bodyChanged bool) {
+func (s *SyncData) IsSGWrite(ctx context.Context, cas uint64, rawBody []byte, rawUserXattr []byte, cv cvExtractor) (isSGWrite bool, crc32Match bool, bodyChanged bool) {
 
 	// If cas matches, it was a SG write
 	if cas == s.GetSyncCas() {
@@ -686,20 +686,26 @@ func (s *SyncData) IsSGWrite(cas uint64, rawBody []byte, rawUserXattr []byte, cv
 		return false, false, false
 	}
 
+	if s.RevAndVersion.CurrentVersion != "" || s.RevAndVersion.CurrentSource != "" {
+		extractedCV, err := cv.ExtractCV()
+		if !errors.Is(err, base.ErrNotFound) {
+			if err != nil {
+				base.WarnfCtx(ctx, "Error extracting cv during IsSGWrite write check: %v", err)
+				return true, true, false
+			}
+			if !s.CVEqual(*extractedCV) {
+				return false, true, false
+			}
+		}
+	}
 	return true, true, false
 }
 
 // IsSGWrite - used during on-demand import. Check SyncData and HLV to determine if the document was written by Sync Gateway or by a Couchbase Server SDK write.
 func (doc *Document) IsSGWrite(ctx context.Context, rawBody []byte) (isSGWrite bool, crc32Match bool, bodyChanged bool) {
-	var cv *Version
-	if doc.RevAndVersion.CurrentSource != "" && doc.RevAndVersion.CurrentVersion != "" && doc.HLV != nil {
-		sourceID, version := doc.HLV.GetCurrentVersion()
-		cv = &Version{SourceID: sourceID, Value: version}
-	}
-
 	// If the raw body is available, use SyncData.IsSGWrite
 	if rawBody != nil && len(rawBody) > 0 {
-		isSgWriteFeed, crc32MatchFeed, bodyChangedFeed := doc.SyncData.IsSGWrite(doc.Cas, rawBody, doc.rawUserXattr, cv)
+		isSgWriteFeed, crc32MatchFeed, bodyChangedFeed := doc.SyncData.IsSGWrite(ctx, doc.Cas, rawBody, doc.rawUserXattr, doc.HLV)
 		return isSgWriteFeed, crc32MatchFeed, bodyChangedFeed
 	}
 
@@ -730,14 +736,14 @@ func (doc *Document) IsSGWrite(ctx context.Context, rawBody []byte) (isSGWrite b
 
 	if HasUserXattrChanged(doc.rawUserXattr, doc.Crc32cUserXattr) {
 		base.DebugfCtx(ctx, base.KeyCRUD, "Doc %s is not an SG write, based on user xattr hash", base.UD(doc.ID))
-		return false, true, false
+		return false, false, false
 	}
-	if doc.RevAndVersion.CurrentSource == "" || doc.RevAndVersion.CurrentVersion == "" {
+	if doc.RevAndVersion.CurrentSource == "" && doc.RevAndVersion.CurrentVersion == "" {
 		return true, true, false
 	}
 
-	if cv != nil {
-		if !doc.CVEqual(cv.SourceID, cv.Value) {
+	if doc.HLV != nil {
+		if !doc.CVEqual(*doc.HLV.ExtractCurrentVersionFromHLV()) {
 			base.DebugfCtx(ctx, base.KeyCRUD, "Doc %s is not an SG write, based on mismatch between version vector cv %s and sync metadata cv %s", base.UD(doc.ID), doc.HLV.GetCurrentVersionString(), doc.RevAndVersion.CV())
 			return false, true, false
 		}

db/hybrid_logical_vector.go

@@ -248,6 +248,14 @@ func (hlv *HybridLogicalVector) ExtractCurrentVersionFromHLV() *Version {
 	return &currVersion
 }
 
+// ExtractCV is used to sastify CV only interface. Since it can never return an error, consider ExtractCurrentVersionFromHLV or GetCurrentVersion instead.
+func (hlv *HybridLogicalVector) ExtractCV() (*Version, error) {
+	if hlv == nil {
+		return nil, base.ErrNotFound
+	}
+	return hlv.ExtractCurrentVersionFromHLV(), nil
+}
+
 // HybridLogicalVector is the in memory format for the hLv.
 type HybridLogicalVector struct {
 	CurrentVersionCAS uint64      // current version cas (or cvCAS) stores the current CAS in little endian hex format at the time of replication
@@ -980,16 +988,29 @@ func GetGenerationFromEncodedVersionValue(value uint64) int {
 	return int((value >> 40) & 0xFFFFFF)
 }
 
-// getCurrentVersionFromVVXattr will extract only the current version from a full version vector xattr, used to
-// optimize unmarshalling when only the current version is needed.
-func getCurrentVersionFromVVXattr(hlvData []byte) (*Version, error) {
+type rawHLV []byte
+
+// GetCurrentVersion returns the current version from the HLV by unmarshalling a raw _vv xattr.
+func (r *rawHLV) ExtractCV() (*Version, error) {
+	if r == nil {
+		return nil, base.ErrNotFound
+	}
 	limitedHLV := struct {
 		Version  string `json:"ver"`
 		SourceID string `json:"src"`
 	}{}
-	err := base.JSONUnmarshal(hlvData, &limitedHLV)
+	err := base.JSONUnmarshal(*r, &limitedHLV)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("could not extract ver and src from _vv xattr (%q): %w", string(*r), err)
 	}
 	return &Version{SourceID: limitedHLV.SourceID, Value: base.HexCasToUint64(limitedHLV.Version)}, nil
 }
+
+// cvExtractor can extract a current version different HLV representations.
+type cvExtractor interface {
+	// ExtractVersion returns the current version from the HLV.
+	ExtractCV() (*Version, error)
+}
+
+var _ cvExtractor = &rawHLV{}
+var _ cvExtractor = &HybridLogicalVector{}

db/import_listener.go

@@ -202,16 +202,13 @@ func (il *importListener) ImportFeedEvent(ctx context.Context, collection *Datab
 			il.importStats.ImportErrorCount.Add(1)
 			return
 		}
-		var cv *Version
+		var cv cvExtractor
 		vv := rawDoc.Xattrs[base.VvXattrName]
 		if len(vv) > 0 {
-			cv, err = getCurrentVersionFromVVXattr(vv)
-			if err != nil {
-				base.WarnfCtx(ctx, "Error parsing version xattr for doc %s - not importing. Continuing without doing version vector consistency check between _vv and _sync: %s", base.UD(event.Key), err)
-				return
-			}
+			cv = base.Ptr(rawHLV(vv))
 		}
-		isSGWrite, crc32Match, _ = syncData.IsSGWrite(event.Cas, rawDoc.Body, rawDoc.Xattrs[collection.userXattrKey()], cv)
+
+		isSGWrite, crc32Match, _ = syncData.IsSGWrite(ctx, event.Cas, rawDoc.Body, rawDoc.Xattrs[collection.userXattrKey()], cv)
 		if crc32Match {
 			il.dbStats.Crc32MatchCount.Add(1)
 		}

xdcr/xdcr_test.go

@@ -690,7 +690,7 @@ func TestXDCRBeforeAttachmentMigration(t *testing.T) {
 	requireWaitForXDCRDocsWritten(t, xdcr, 3)
 
 	// check that we can fetch the attachment now
-	// Due to CBG-4864, this writes a new revision since _vv.ver > _sync.rev.ver even though the body didn't change.
+	// Due to CBG-4878, this writes a new revision since _vv.ver > _sync.rev.ver even though the body didn't change.
 	// This isn't ideal and it would be better to not have this issue a new revision.
 	//
 	// Note: The versions remaining the same means that clients that observe the missing attachment never get it until a subsequent doc update
@@ -705,7 +705,7 @@ func TestXDCRBeforeAttachmentMigration(t *testing.T) {
 	attachmentsAfter := db.GetRawGlobalSyncAttachments(t, dstDs, docID)
 	require.Contains(t, attachmentsAfter, attName)
 
-	// CBG-4864 causes these to not be equal but equality would be better.
+	// CBG-4878 causes these to not be equal but equality would be better.
 	assert.NotEqual(t, dstPreMigrateSeq, dstPostMigrateSeq)
 	assert.NotEqual(t, dstPreMigrateRev, dstPostMigrateRev)
 }