CBG-4420: handle rev tree in history on processRev

.github/workflows/ci.yml

@@ -134,25 +134,6 @@ jobs:
         with:
           test-results: test.json
 
-  test-no-xattrs:
-    runs-on: ubuntu-latest
-    env:
-      GOPRIVATE: github.com/couchbaselabs
-      SG_TEST_USE_XATTRS: false
-    steps:
-      - uses: actions/checkout@v4
-      - uses: actions/setup-go@v5
-        with:
-          go-version: 1.23.3
-      - name: Run Tests
-        run: go test -tags cb_sg_devmode -shuffle=on -timeout=30m -count=1 -json -v "./..." | tee test.json | jq -s -jr 'sort_by(.Package,.Time) | .[].Output | select (. != null )'
-        shell: bash
-      - name: Annotate Failures
-        if: always()
-        uses: guyarb/golang-test-annotations@v0.8.0
-        with:
-          test-results: test.json
-
   python-format:
     runs-on: ubuntu-latest
     steps:

.golangci.yml

@@ -73,6 +73,7 @@ issues:
     - path: (_test\.go|utilities_testing\.go)
       linters:
         - goconst
+        - prealloc
     - path: (_test\.go|utilities_testing\.go)
       linters:
         - govet

base/bucket_gocb_test.go

@@ -414,11 +414,11 @@ func TestXattrWriteCasSimple(t *testing.T) {
 	assert.Equal(t, Crc32cHashString(valBytes), macroBodyHashString)
 
 	// Validate against $document.value_crc32c
-	_, xattrs, _, err = dataStore.GetWithXattrs(ctx, key, []string{"$document"})
+	_, xattrs, _, err = dataStore.GetWithXattrs(ctx, key, []string{VirtualDocumentXattr})
 	require.NoError(t, err)
 
 	var retrievedVxattr map[string]interface{}
-	require.NoError(t, json.Unmarshal(xattrs["$document"], &retrievedVxattr))
+	require.NoError(t, json.Unmarshal(xattrs[VirtualDocumentXattr], &retrievedVxattr))
 
 	vxattrCrc32c, ok := retrievedVxattr["value_crc32c"].(string)
 	assert.True(t, ok, "Unable to retrieve virtual xattr crc32c as string")

base/constants.go

@@ -135,14 +135,21 @@ const (
 	// SyncPropertyName is used when storing sync data inline in a document.
 	SyncPropertyName = "_sync"
 	// SyncXattrName is used when storing sync data in a document's xattrs.
-	SyncXattrName = "_sync"
+	SyncXattrName   = "_sync"
+	VvXattrName     = "_vv"
+	GlobalXattrName = "_globalSync"
 
 	// MouXattrName is used when storing metadata-only update information in a document's xattrs.
 	MouXattrName = "_mou"
 
 	// Intended to be used in Meta Map and related tests
 	MetaMapXattrsKey = "xattrs"
 
+	// VirtualXattrRevSeqNo is used to fetch rev seq no from documents virtual xattr
+	VirtualXattrRevSeqNo = "$document.revid"
+	// VirtualDocumentXattr is used to fetch the documents virtual xattr
+	VirtualDocumentXattr = "$document"
+
 	// Prefix for transaction metadata documents
 	TxnPrefix = "_txn:"
 

base/constants_syncdocs.go

@@ -11,6 +11,7 @@ licenses/APL2.txt.
 package base
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"strconv"
@@ -377,58 +378,133 @@ func CollectionSyncFunctionKeyWithGroupID(groupID string, scopeName, collectionN
 
 // SyncInfo documents are stored in collections to identify the metadataID associated with sync metadata in that collection
 type SyncInfo struct {
-	MetadataID string `json:"metadataID"`
+	MetadataID      string `json:"metadataID,omitempty"`
+	MetaDataVersion string `json:"metadata_version,omitempty"`
 }
 
 // initSyncInfo attempts to initialize syncInfo for a datastore
 //  1. If syncInfo doesn't exist, it is created for the specified metadataID
 //  2. If syncInfo exists with a matching metadataID, returns requiresResync=false
 //  3. If syncInfo exists with a non-matching metadataID, returns requiresResync=true
-func InitSyncInfo(ds DataStore, metadataID string) (requiresResync bool, err error) {
+//     If syncInfo exists and has metaDataVersion greater than or equal to 4.0, return requiresAttachmentMigration=false, else requiresAttachmentMigration=true to bring migrate metadata attachments.
+func InitSyncInfo(ctx context.Context, ds DataStore, metadataID string) (requiresResync bool, requiresAttachmentMigration bool, err error) {
 
 	var syncInfo SyncInfo
 	_, fetchErr := ds.Get(SGSyncInfo, &syncInfo)
 	if IsDocNotFoundError(fetchErr) {
 		if metadataID == "" {
-			return false, nil
+			return false, true, nil
 		}
 		newSyncInfo := &SyncInfo{MetadataID: metadataID}
 		_, addErr := ds.Add(SGSyncInfo, 0, newSyncInfo)
 		if IsCasMismatch(addErr) {
 			// attempt new fetch
 			_, fetchErr = ds.Get(SGSyncInfo, &syncInfo)
 			if fetchErr != nil {
-				return true, fmt.Errorf("Error retrieving syncInfo (after failed add): %v", fetchErr)
+				return true, true, fmt.Errorf("Error retrieving syncInfo (after failed add): %v", fetchErr)
 			}
 		} else if addErr != nil {
-			return true, fmt.Errorf("Error adding syncInfo: %v", addErr)
+			return true, true, fmt.Errorf("Error adding syncInfo: %v", addErr)
 		}
 		// successfully added
-		return false, nil
+		requiresAttachmentMigration, err = CompareMetadataVersion(ctx, syncInfo.MetaDataVersion)
+		if err != nil {
+			return syncInfo.MetadataID != metadataID, true, err
+		}
+		return false, requiresAttachmentMigration, nil
 	} else if fetchErr != nil {
-		return true, fmt.Errorf("Error retrieving syncInfo: %v", fetchErr)
+		return true, true, fmt.Errorf("Error retrieving syncInfo: %v", fetchErr)
+	}
+	// check for meta version, if we don't have meta version of 4.0 we need to run migration job
+	requiresAttachmentMigration, err = CompareMetadataVersion(ctx, syncInfo.MetaDataVersion)
+	if err != nil {
+		return syncInfo.MetadataID != metadataID, true, err
 	}
 
-	return syncInfo.MetadataID != metadataID, nil
+	return syncInfo.MetadataID != metadataID, requiresAttachmentMigration, nil
 }
 
-// SetSyncInfo sets syncInfo in a DataStore to the specified metadataID
-func SetSyncInfo(ds DataStore, metadataID string) error {
+// SetSyncInfoMetadataID sets syncInfo in a DataStore to the specified metadataID, preserving metadata version if present
+func SetSyncInfoMetadataID(ds DataStore, metadataID string) error {
 
 	// If the metadataID isn't defined, don't persist SyncInfo.  Defensive handling for legacy use cases.
 	if metadataID == "" {
 		return nil
 	}
-	syncInfo := &SyncInfo{
-		MetadataID: metadataID,
+	_, err := ds.Update(SGSyncInfo, 0, func(current []byte) (updated []byte, expiry *uint32, delete bool, err error) {
+		var syncInfo SyncInfo
+		if current != nil {
+			parseErr := JSONUnmarshal(current, &syncInfo)
+			if parseErr != nil {
+				return nil, nil, false, parseErr
+			}
+		}
+		// if we have a metadataID to set, set it preserving the metadata version if present
+		syncInfo.MetadataID = metadataID
+		bytes, err := JSONMarshal(&syncInfo)
+		return bytes, nil, false, err
+	})
+	return err
+}
+
+// SetSyncInfoMetaVersion sets sync info in DataStore to specified metadata version, preserving metadataID if present
+func SetSyncInfoMetaVersion(ds DataStore, metaVersion string) error {
+	if metaVersion == "" {
+		return nil
 	}
-	return ds.Set(SGSyncInfo, 0, nil, syncInfo)
+	_, err := ds.Update(SGSyncInfo, 0, func(current []byte) (updated []byte, expiry *uint32, delete bool, err error) {
+		var syncInfo SyncInfo
+		if current != nil {
+			parseErr := JSONUnmarshal(current, &syncInfo)
+			if parseErr != nil {
+				return nil, nil, false, parseErr
+			}
+		}
+		// if we have a meta version to set, set it preserving the metadata ID if present
+		syncInfo.MetaDataVersion = metaVersion
+		bytes, err := JSONMarshal(&syncInfo)
+		return bytes, nil, false, err
+	})
+	return err
 }
 
-// SerializeIfLonger returns name as a sha1 string if the length of the name is greater or equal to the length specificed. Otherwise, returns the original string.
+// SerializeIfLonger returns name as a sha1 string if the length of the name is greater or equal to the length specified. Otherwise, returns the original string.
 func SerializeIfLonger(name string, length int) string {
 	if len(name) < length {
 		return name
 	}
 	return Sha1HashString(name, "")
 }
+
+// CompareMetadataVersion Will build comparable build version for comparison with meta version defined in syncInfo, then
+// will return true if we require attachment migration, false if not.
+func CompareMetadataVersion(ctx context.Context, metaVersion string) (bool, error) {
+	if metaVersion == "" {
+		// no meta version passed in, thus attachment migration should take place
+		return true, nil
+	}
+	syncInfoVersion, err := NewComparableBuildVersionFromString(metaVersion)
+	if err != nil {
+		return true, err
+	}
+	return CheckRequireAttachmentMigration(ctx, syncInfoVersion)
+}
+
+// CheckRequireAttachmentMigration will return true if current metaVersion < 4.0.0, else false
+func CheckRequireAttachmentMigration(ctx context.Context, version *ComparableBuildVersion) (bool, error) {
+	if version == nil {
+		AssertfCtx(ctx, "failed to build comparable build version for syncInfo metaVersion")
+		return true, fmt.Errorf("corrupt syncInfo metaVersion value")
+	}
+	minVerStr := "4.0.0" // minimum meta version that needs to be defined for metadata migration. Any version less than this will require attachment migration
+	minVersion, err := NewComparableBuildVersionFromString(minVerStr)
+	if err != nil {
+		AssertfCtx(ctx, "failed to build comparable build version for minimum version for attachment migration")
+		return true, err
+	}
+
+	if minVersion.AtLeastMinorDowngrade(version) {
+		return true, nil
+	}
+	return false, nil
+}

base/dcp_common.go

@@ -522,12 +522,13 @@ func dcpKeyFilter(key []byte, metaKeys *MetadataKeys) bool {
 }
 
 // Makes a feedEvent that can be passed to a FeedEventCallbackFunc implementation
-func makeFeedEvent(key []byte, value []byte, dataType uint8, cas uint64, expiry uint32, vbNo uint16, collectionID uint32, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
+func makeFeedEvent(key []byte, value []byte, dataType uint8, cas uint64, expiry uint32, vbNo uint16, collectionID uint32, revNo uint64, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
 
 	// not currently doing rq.Extras handling (as in gocouchbase/upr_feed, makeUprEvent) as SG doesn't use
 	// expiry/flags information, and snapshot handling is done by cbdatasource and sent as
 	// SnapshotStart, SnapshotEnd
 	event := sgbucket.FeedEvent{
+		RevNo:        revNo,
 		Opcode:       opcode,
 		Key:          key,
 		Value:        value,

base/dcp_dest.go

@@ -92,7 +92,7 @@ func (d *DCPDest) DataUpdate(partition string, key []byte, seq uint64,
 	if !dcpKeyFilter(key, d.metaKeys) {
 		return nil
 	}
-	event := makeFeedEventForDest(key, val, cas, partitionToVbNo(d.loggingCtx, partition), collectionIDFromExtras(extras), 0, 0, sgbucket.FeedOpMutation)
+	event := makeFeedEventForDest(key, val, cas, partitionToVbNo(d.loggingCtx, partition), collectionIDFromExtras(extras), 0, 0, 0, sgbucket.FeedOpMutation)
 	d.dataUpdate(seq, event)
 	return nil
 }
@@ -116,7 +116,7 @@ func (d *DCPDest) DataUpdateEx(partition string, key []byte, seq uint64, val []b
 		if !ok {
 			return errors.New("Unable to cast extras of type DEST_EXTRAS_TYPE_GOCB_DCP to cbgt.GocbExtras")
 		}
-		event = makeFeedEventForDest(key, val, cas, partitionToVbNo(d.loggingCtx, partition), dcpExtras.CollectionId, dcpExtras.Expiry, dcpExtras.Datatype, sgbucket.FeedOpMutation)
+		event = makeFeedEventForDest(key, val, cas, partitionToVbNo(d.loggingCtx, partition), dcpExtras.CollectionId, dcpExtras.Expiry, dcpExtras.Datatype, dcpExtras.RevNo, sgbucket.FeedOpMutation)
 
 	}
 
@@ -131,7 +131,7 @@ func (d *DCPDest) DataDelete(partition string, key []byte, seq uint64,
 		return nil
 	}
 
-	event := makeFeedEventForDest(key, nil, cas, partitionToVbNo(d.loggingCtx, partition), collectionIDFromExtras(extras), 0, 0, sgbucket.FeedOpDeletion)
+	event := makeFeedEventForDest(key, nil, cas, partitionToVbNo(d.loggingCtx, partition), collectionIDFromExtras(extras), 0, 0, 0, sgbucket.FeedOpDeletion)
 	d.dataUpdate(seq, event)
 	return nil
 }
@@ -154,7 +154,7 @@ func (d *DCPDest) DataDeleteEx(partition string, key []byte, seq uint64,
 		if !ok {
 			return errors.New("Unable to cast extras of type DEST_EXTRAS_TYPE_GOCB_DCP to cbgt.GocbExtras")
 		}
-		event = makeFeedEventForDest(key, dcpExtras.Value, cas, partitionToVbNo(d.loggingCtx, partition), dcpExtras.CollectionId, dcpExtras.Expiry, dcpExtras.Datatype, sgbucket.FeedOpDeletion)
+		event = makeFeedEventForDest(key, dcpExtras.Value, cas, partitionToVbNo(d.loggingCtx, partition), dcpExtras.CollectionId, dcpExtras.Expiry, dcpExtras.Datatype, dcpExtras.RevNo, sgbucket.FeedOpDeletion)
 
 	}
 	d.dataUpdate(seq, event)
@@ -247,8 +247,8 @@ func collectionIDFromExtras(extras []byte) uint32 {
 	return binary.LittleEndian.Uint32(extras[4:])
 }
 
-func makeFeedEventForDest(key []byte, val []byte, cas uint64, vbNo uint16, collectionID uint32, expiry uint32, dataType uint8, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
-	return makeFeedEvent(key, val, dataType, cas, expiry, vbNo, collectionID, opcode)
+func makeFeedEventForDest(key []byte, val []byte, cas uint64, vbNo uint16, collectionID uint32, expiry uint32, dataType uint8, revNo uint64, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
+	return makeFeedEvent(key, val, dataType, cas, expiry, vbNo, collectionID, revNo, opcode)
 }
 
 // DCPLoggingDest wraps DCPDest to provide per-callback logging

base/dcp_feed_type.go

@@ -296,7 +296,7 @@ func getCbgtCredentials(dbName string) (cbgtCreds, bool) {
 	return creds, found
 }
 
-// See the comment of cbgtRootCAsProvider for usage details.
+// setCbgtRootCertsForBucket creates root certificates for a given bucket. If TLS should be used, this function must be called. If tls certificate verification is skipped, then this function should be called with pool as nil. See the comment of cbgtRootCAsProvider for usage details.
 func setCbgtRootCertsForBucket(bucketUUID string, pool *x509.CertPool) {
 	cbgtGlobalsLock.Lock()
 	defer cbgtGlobalsLock.Unlock()

base/dcp_receiver.go

@@ -26,7 +26,7 @@ const MemcachedDataTypeRaw = 0
 
 // Make a feed event for a gomemcached request.  Extracts expiry from extras
 func makeFeedEventForMCRequest(rq *gomemcached.MCRequest, opcode sgbucket.FeedOpcode) sgbucket.FeedEvent {
-	return makeFeedEvent(rq.Key, rq.Body, rq.DataType, rq.Cas, ExtractExpiryFromDCPMutation(rq), rq.VBucket, 0, opcode)
+	return makeFeedEvent(rq.Key, rq.Body, rq.DataType, rq.Cas, ExtractExpiryFromDCPMutation(rq), rq.VBucket, 0, 0, opcode)
 }
 
 // ShardedImportDCPMetadata is an internal struct that is exposed to enable json marshaling, used by sharded import feed. It differs from DCPMetadata because it must match the private struct used by cbgt.metadata.

base/dcp_sharded.go

@@ -451,6 +451,7 @@ func (c *CbgtContext) Stop() {
 
 func (c *CbgtContext) RemoveFeedCredentials(dbName string) {
 	removeCbgtCredentials(dbName)
+	// CBG-4394: removing root certs for the bucket should be done, but it is keyed based on the bucket UUID, and multiple dbs can use the same bucket
 }
 
 // Format of dest key for retrieval of import dest from cbgtDestFactories

base/log_keys.go

@@ -53,6 +53,7 @@ const (
 	KeyReplicate
 	KeySync
 	KeySyncMsg
+	KeyVV
 	KeyWebSocket
 	KeyWebSocketFrame
 	KeySGTest
@@ -87,6 +88,7 @@ var (
 		KeyReplicate:      "Replicate",
 		KeySync:           "Sync",
 		KeySyncMsg:        "SyncMsg",
+		KeyVV:             "VV",
 		KeyWebSocket:      "WS",
 		KeyWebSocketFrame: "WSFrame",
 		KeySGTest:         "TEST",

base/main_test_bucket_pool.go

@@ -137,6 +137,15 @@ func NewTestBucketPoolWithOptions(ctx context.Context, bucketReadierFunc TBPBuck
 	}
 	tbp.skipMobileXDCR = !useMobileXDCR
 
+	// at least anemone release
+	if os.Getenv(tbpEnvAllowIncompatibleServerVersion) == "" && !ProductVersion.Less(&ComparableBuildVersion{major: 4}) {
+		overrideMsg := "Set " + tbpEnvAllowIncompatibleServerVersion + "=true to override this check."
+		// this check also covers BucketStoreFeatureMultiXattrSubdocOperations, which is Couchbase Server 7.6
+		if tbp.skipMobileXDCR {
+			tbp.Fatalf(ctx, "Sync Gateway %v requires mobile XDCR support, but Couchbase Server %v does not support it. Couchbase Server %s is required. %s", ProductVersion, tbp.cluster.version, firstServerVersionToSupportMobileXDCR, overrideMsg)
+		}
+	}
+
 	tbp.verbose.Set(tbpVerbose())
 
 	// Start up an async readier worker to process dirty buckets
@@ -450,6 +459,7 @@ func (tbp *TestBucketPool) setXDCRBucketSetting(ctx context.Context, bucket Buck
 
 	tbp.Logf(ctx, "Setting crossClusterVersioningEnabled=true")
 
+	// retry for 1 minute to get this bucket setting, MB-63675
 	store, ok := AsCouchbaseBucketStore(bucket)
 	if !ok {
 		tbp.Fatalf(ctx, "unable to get server management endpoints. Underlying bucket type was not GoCBBucket")
@@ -459,12 +469,20 @@ func (tbp *TestBucketPool) setXDCRBucketSetting(ctx context.Context, bucket Buck
 	posts.Add("enableCrossClusterVersioning", "true")
 
 	url := fmt.Sprintf("/pools/default/buckets/%s", store.GetName())
-	output, statusCode, err := store.MgmtRequest(ctx, http.MethodPost, url, "application/x-www-form-urlencoded", strings.NewReader(posts.Encode()))
+	// retry for 1 minute to get this bucket setting, MB-63675
+	_, err := RetryLoop(ctx, "setXDCRBucketSetting", func() (bool, error, interface{}) {
+		output, statusCode, err := store.MgmtRequest(ctx, http.MethodPost, url, "application/x-www-form-urlencoded", strings.NewReader(posts.Encode()))
+		if err != nil {
+			tbp.Fatalf(ctx, "request to mobile XDCR bucket setting failed, status code: %d error: %w output: %s", statusCode, err, string(output))
+		}
+		if statusCode != http.StatusOK {
+			err := fmt.Errorf("request to mobile XDCR bucket setting failed with status code, %d, output: %s", statusCode, string(output))
+			return true, err, nil
+		}
+		return false, nil, nil
+	}, CreateMaxDoublingSleeperFunc(200, 500, 500))
 	if err != nil {
-		tbp.Fatalf(ctx, "request to mobile XDCR bucket setting failed, status code: %d error: %v output: %s", statusCode, err, string(output))
-	}
-	if statusCode != http.StatusOK {
-		tbp.Fatalf(ctx, "request to mobile XDCR bucket setting failed with status code, %d, output: %s", statusCode, string(output))
+		tbp.Fatalf(ctx, "Couldn't set crossClusterVersioningEnabled: %v", err)
 	}
 }
 

base/main_test_bucket_pool_config.go

@@ -53,11 +53,17 @@ const (
 
 	tbpEnvUseDefaultCollection = "SG_TEST_USE_DEFAULT_COLLECTION"
 
+	// tbpEnvAllowIncompatibleServerVersion allows tests to run against a server version that is not presumed compatible with version of Couchbase Server running.
+	tbpEnvAllowIncompatibleServerVersion = "SG_TEST_SKIP_SERVER_VERSION_CHECK"
+
 	// wait this long when requesting a test bucket from the pool before giving up and failing the test.
 	waitForReadyBucketTimeout = time.Minute
 
 	// Creates buckets with a specific number of number of replicas
 	tbpEnvBucketNumReplicas = "SG_TEST_BUCKET_NUM_REPLICAS"
+
+	// Environment variable to specify the topology tests to run
+	TbpEnvTopologyTests = "SG_TEST_TOPOLOGY_TESTS"
 )
 
 // TestsUseNamedCollections returns true if the tests use named collections.