CBG-4884: Skip removal of obsolete attachments if ECCV is set (#7783)

Co-authored-by: Tor Colvin <[email protected]>

Author: bbrks

db/crud.go

@@ -2685,6 +2685,11 @@ func (db *DatabaseCollectionWithUser) updateAndReturnDoc(ctx context.Context, do
 	skipObsoleteAttachmentsRemoval := false
 	isNewDocCreation := false
 
+	// Don't remove obsolete attachments if using ECCV - the other cluster may still need them!
+	if db.dbCtx.CachedCCVEnabled.Load() {
+		skipObsoleteAttachmentsRemoval = true
+	}
+
 	if db.UseXattrs() || upgradeInProgress {
 		var casOut uint64
 		// Update the document, storing metadata in extended attribute

rest/attachment_test.go

@@ -2055,60 +2055,76 @@ func TestAttachmentRemovalWithConflicts(t *testing.T) {
 	rt := NewRestTester(t, nil)
 
 	defer rt.Close()
-
 	rt.GetDatabase().EnableAllowConflicts(rt.TB())
-	const docID = "doc"
-	// Create doc rev 1
-	version := rt.PutDoc(docID, `{"test": "x"}`)
 
-	// Create doc rev 2 with attachment
-	version = rt.UpdateDoc(docID, version, `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
+	testCases := []struct {
+		name string
+		eccv bool
+	}{
+		{name: "no eccv", eccv: false},
+		{name: "eccv", eccv: true},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			rt.GetDatabase().CachedCCVEnabled.Store(tc.eccv)
 
-	// Create doc rev 3 referencing previous attachment
-	losingVersion3 := rt.UpdateDoc(docID, version, `{"_attachments": {"hello.txt": {"revpos":2,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`)
+			docID := db.SafeDocumentName(t, t.Name())
+			// Create doc rev 1
+			version := rt.PutDoc(docID, `{"test": "x"}`)
 
-	// Create doc conflicting with previous revid referencing previous attachment too
-	winningVersion3 := rt.PutNewEditsFalse(docID, NewDocVersionFromFakeRev("3-b"), &version, `{"_attachments": {"hello.txt": {"revpos":2,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}, "Winning Rev": true}`)
+			// Create doc rev 2 with attachment
+			version = rt.UpdateDoc(docID, version, `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}}`)
 
-	// Update the winning rev 3 and ensure attachment remains around as the other leaf still references this attachment
-	finalVersion4 := rt.UpdateDoc(docID, *winningVersion3, `{"update": 2}`)
+			// Create doc rev 3 referencing previous attachment
+			losingVersion3 := rt.UpdateDoc(docID, version, `{"_attachments": {"hello.txt": {"revpos":2,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}}`)
 
-	type docResp struct {
-		Attachments db.AttachmentMap `json:"_attachments"`
-	}
+			// Create doc conflicting with previous revid referencing previous attachment too
+			winningVersion3 := rt.PutNewEditsFalse(docID, NewDocVersionFromFakeRev("3-b"), &version, `{"_attachments": {"hello.txt": {"revpos":2,"stub":true,"digest":"sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0="}}, "Winning Rev": true}`)
 
-	var doc1 docResp
-	// Get losing rev and ensure attachment is still there and has not been deleted
-	resp := rt.SendAdminRequestWithHeaders("GET", "/{{.keyspace}}/doc?attachments=true&rev="+losingVersion3.RevTreeID, "", map[string]string{"Accept": "application/json"})
-	RequireStatus(t, resp, http.StatusOK)
+			// Update the winning rev 3 and ensure attachment remains around as the other leaf still references this attachment
+			finalVersion4 := rt.UpdateDoc(docID, *winningVersion3, `{"update": 2}`)
 
-	err := base.JSONUnmarshal(resp.BodyBytes(), &doc1)
-	assert.NoError(t, err)
-	require.Contains(t, doc1.Attachments, "hello.txt")
-	require.Equal(t, db.DocAttachment{
-		Digest: "sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0=",
-		Length: 11,
-		Revpos: 2,
-		Data:   []byte("hello world"),
-	}, doc1.Attachments["hello.txt"])
-
-	attachmentKey := db.MakeAttachmentKey(2, "doc", doc1.Attachments["hello.txt"].Digest)
-
-	var doc2 docResp
-	// Get winning rev and ensure attachment is indeed removed from this rev
-	resp = rt.SendAdminRequestWithHeaders("GET", "/{{.keyspace}}/doc?attachments=true&rev="+finalVersion4.RevTreeID, "", map[string]string{"Accept": "application/json"})
-	RequireStatus(t, resp, http.StatusOK)
+			type docResp struct {
+				Attachments db.AttachmentMap `json:"_attachments"`
+			}
 
-	err = base.JSONUnmarshal(resp.BodyBytes(), &doc2)
-	assert.NoError(t, err)
-	require.NotContains(t, doc2.Attachments, "hello.txt")
+			var doc1 docResp
+			// Get losing rev and ensure attachment is still there and has not been deleted
+			resp := rt.SendAdminRequestWithHeaders("GET", "/{{.keyspace}}/"+docID+"?attachments=true&rev="+losingVersion3.RevTreeID, "", map[string]string{"Accept": "application/json"})
+			RequireStatus(t, resp, http.StatusOK)
+
+			err := base.JSONUnmarshal(resp.BodyBytes(), &doc1)
+			assert.NoError(t, err)
+			require.Contains(t, doc1.Attachments, "hello.txt")
+			require.Equal(t, db.DocAttachment{
+				Digest: "sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0=",
+				Length: 11,
+				Revpos: 2,
+				Data:   []byte("hello world"),
+			}, doc1.Attachments["hello.txt"])
+
+			attachmentKey := db.MakeAttachmentKey(2, docID, doc1.Attachments["hello.txt"].Digest)
+
+			var doc2 docResp
+			// Get winning rev and ensure attachment is indeed removed from this rev
+			resp = rt.SendAdminRequestWithHeaders("GET", "/{{.keyspace}}/"+docID+"?attachments=true&rev="+finalVersion4.RevTreeID, "", map[string]string{"Accept": "application/json"})
+			RequireStatus(t, resp, http.StatusOK)
+
+			err = base.JSONUnmarshal(resp.BodyBytes(), &doc2)
+			assert.NoError(t, err)
+			require.NotContains(t, doc2.Attachments, "hello.txt")
 
-	// Now remove the attachment in the losing rev by deleting the revision and ensure the attachment gets deleted
-	rt.DeleteDoc(docID, losingVersion3)
+			// Now remove the attachment in the losing rev by deleting the revision and ensure the attachment gets deleted
+			rt.DeleteDoc(docID, losingVersion3)
 
-	_, _, err = rt.GetSingleDataStore().GetRaw(attachmentKey)
-	assert.Error(t, err)
-	assert.True(t, base.IsDocNotFoundError(err))
+			_, _, err = rt.GetSingleDataStore().GetRaw(attachmentKey)
+			if rt.GetDatabase().CachedCCVEnabled.Load() {
+				require.NoError(t, err, "Attachment should not be deleted as eccv is enabled")
+			} else {
+				base.RequireDocNotFoundError(t, err)
+			}
+		})
+	}
 }
 
 func TestAttachmentsMissing(t *testing.T) {
@@ -2208,36 +2224,45 @@ func TestAttachmentDeleteOnExpiry(t *testing.T) {
 
 	dataStore := rt.GetSingleDataStore()
 
-	// Create doc with attachment and expiry
-	resp := rt.SendAdminRequest("PUT", "/{{.keyspace}}/"+t.Name(), `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}, "_exp": 1}`)
-	RequireStatus(t, resp, http.StatusCreated)
-
-	// Wait for document to be expired - this bucket get should also trigger the expiry purge interval
-	require.EventuallyWithT(t, func(c *assert.CollectT) {
-		_, _, err := dataStore.GetRaw(t.Name())
-		assert.True(c, base.IsDocNotFoundError(err), "expected err %v to be doc not found", err)
-	}, time.Second*10, time.Millisecond*10)
-
-	if base.TestUseXattrs() {
-		require.EventuallyWithT(t, func(c *assert.CollectT) {
-			assert.Equal(c, int64(1), rt.GetDatabase().DbStats.SharedBucketImport().ImportCount.Value())
-		}, time.Second*10, time.Millisecond*5)
-	} else {
-		// Trigger OnDemand Import for that doc to trigger tombstone
-		resp := rt.SendAdminRequest("GET", "/{{.keyspace}}/"+t.Name(), "")
-		RequireStatus(t, resp, http.StatusNotFound)
+	testCases := []struct {
+		name string
+		eccv bool
+	}{
+		{name: "no-eccv", eccv: false},
+		{name: "eccv", eccv: true},
 	}
-	att2Key := db.MakeAttachmentKey(db.AttVersion2, t.Name(), "sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0=")
-
-	// With xattrs doc will be imported and will be captured as tombstone and therefore purge attachments
-	// Otherwise attachment will not be purged
-	_, _, err := dataStore.GetRaw(att2Key)
-	if base.TestUseXattrs() {
-		base.RequireDocNotFoundError(t, err)
-	} else {
-		assert.NoError(t, err)
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			lastImportCount := rt.GetDatabase().DbStats.SharedBucketImport().ImportCount.Value()
+
+			rt.GetDatabase().CachedCCVEnabled.Store(tc.eccv)
+			docID := db.SafeDocumentName(t, t.Name())
+			// Create doc with attachment and expiry
+			resp := rt.SendAdminRequest("PUT", "/{{.keyspace}}/"+docID, `{"_attachments": {"hello.txt": {"data": "aGVsbG8gd29ybGQ="}}, "_exp": 1}`)
+			RequireStatus(t, resp, http.StatusCreated)
+
+			// Wait for document to be expired - this bucket get should also trigger the expiry purge interval
+			require.EventuallyWithT(t, func(c *assert.CollectT) {
+				_, _, err := dataStore.GetRaw(docID)
+				assert.True(c, base.IsDocNotFoundError(err), "expected err %v to be doc not found", err)
+			}, time.Second*10, time.Millisecond*10)
+
+			require.EventuallyWithT(t, func(c *assert.CollectT) {
+				newImportCount := rt.GetDatabase().DbStats.SharedBucketImport().ImportCount.Value() - lastImportCount
+				assert.Equal(c, int64(1), newImportCount)
+			}, time.Second*10, time.Millisecond*5)
+			att2Key := db.MakeAttachmentKey(db.AttVersion2, docID, "sha1-Kq5sNclPz7QV2+lfQIuc6R7oRu0=")
+
+			// With xattrs doc will be imported and will be captured as tombstone and therefore purge attachments
+			// Otherwise attachment will not be purged
+			_, _, err := dataStore.GetRaw(att2Key)
+			if rt.GetDatabase().CachedCCVEnabled.Load() {
+				assert.NoError(t, err)
+			} else {
+				base.RequireDocNotFoundError(t, err)
+			}
+		})
 	}
-
 }
 
 // TestUpdateViaBlipMigrateAttachment:

rest/audit_test.go

@@ -1126,7 +1126,10 @@ func TestAuditAttachmentEvents(t *testing.T) {
 			requireAttachmentEvents(rt, base.AuditIDAttachmentCreate, output, docID, postAttachmentVersion.RevTreeID, attachmentName, testCase.attachmentCreateCount)
 			requireAttachmentEvents(rt, base.AuditIDAttachmentRead, output, docID, postAttachmentVersion.RevTreeID, attachmentName, testCase.attachmentReadCount)
 			requireAttachmentEvents(rt, base.AuditIDAttachmentUpdate, output, docID, postAttachmentVersion.RevTreeID, attachmentName, testCase.attachmentUpdateCount)
-			requireAttachmentEvents(rt, base.AuditIDAttachmentDelete, output, docID, postAttachmentVersion.RevTreeID, attachmentName, testCase.attachmentDeleteCount)
+			// attachments get auto-deleted only without CCV
+			if !rt.GetDatabase().CachedCCVEnabled.Load() {
+				requireAttachmentEvents(rt, base.AuditIDAttachmentDelete, output, docID, postAttachmentVersion.RevTreeID, attachmentName, testCase.attachmentDeleteCount)
+			}
 
 		})
 	}

xdcr/attachment_test.go

@@ -0,0 +1,154 @@
+// Copyright 2025-Present Couchbase, Inc.
+//
+// Use of this software is governed by the Business Source License included
+// in the file licenses/BSL-Couchbase.txt.  As of the Change Date specified
+// in that file, in accordance with the Business Source License, use of this
+// software will be governed by the Apache License, Version 2.0, included in
+// the file licenses/APL2.txt.
+
+package xdcr
+
+import (
+	"encoding/base64"
+	"net/http"
+	"testing"
+	"time"
+
+	"github.com/couchbase/sync_gateway/base"
+	"github.com/couchbase/sync_gateway/rest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestMultiActorLosingConflictUpdateRemovingAttachments
+// Removes attachments on a losing conflicting document and ensures that attachments remain present in the bucket and accessible on both documents on either side.
+//
+// 1. Create a document with an attachment on Actor A
+// 2. Replicate to Actor B
+// 3. Stop replications
+// 4. Update the document on Actor A, removing the attachment
+// 5. Update the document on Actor B, changing the body (twice to ensure MWW resolves this as the winner as well as LWW)
+// 6. Start replications
+// 7. Observe resolved conflict on both Actor A and Actor B, with the attachment still present
+func TestMultiActorLosingConflictUpdateRemovingAttachments(t *testing.T) {
+	base.RequireNumTestBuckets(t, 2)
+
+	base.SetUpTestLogging(t, base.LevelDebug, base.KeyAll)
+
+	// turn off auto import - since we want reliable XDCR stats and don't want MOU/import echos to interfere
+	rtA := rest.NewRestTester(t, &rest.RestTesterConfig{AutoImport: base.Ptr(false)})
+	defer rtA.Close()
+	rtB := rest.NewRestTester(t, &rest.RestTesterConfig{AutoImport: base.Ptr(false)})
+	defer rtB.Close()
+
+	ctx := base.TestCtx(t)
+	opts := XDCROptions{Mobile: MobileOn}
+
+	// Set up bi-directional XDCR
+	xdcrAtoB, err := NewXDCR(ctx, rtA.Bucket(), rtB.Bucket(), opts)
+	require.NoError(t, err)
+	require.NoError(t, xdcrAtoB.Start(ctx))
+	xdcrBtoA, err := NewXDCR(ctx, rtB.Bucket(), rtA.Bucket(), opts)
+	require.NoError(t, err)
+	require.NoError(t, xdcrBtoA.Start(ctx))
+
+	defer func() {
+		// stop XDCR, will already be stopped if test doesn't fail early
+		if err := xdcrAtoB.Stop(ctx); err != nil {
+			assert.Equal(t, ErrReplicationNotRunning, err)
+		}
+		if err := xdcrBtoA.Stop(ctx); err != nil {
+			assert.Equal(t, ErrReplicationNotRunning, err)
+		}
+	}()
+
+	const (
+		docID        = "doc1"
+		attachmentID = "hello.txt"
+	)
+	attachment := base64.StdEncoding.EncodeToString([]byte("Hello World!"))
+
+	rtAVersion := rtA.PutDocWithAttachment(docID, `{"key":"value"}`, attachmentID, attachment)
+
+	// fetch attachment via REST API
+	attAResp := rtA.SendAdminRequest(http.MethodGet, "/{{.keyspace}}/"+docID+"/"+attachmentID, "")
+	rest.RequireStatus(t, attAResp, http.StatusOK)
+
+	// wait for doc to replicate to rtB
+	var rtBVersion rest.DocVersion
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		rtBVersion, _ = rtB.GetDoc(docID)
+		assert.Equal(c, rtAVersion.CV.String(), rtBVersion.CV.String())
+		assert.Equal(c, rtAVersion.RevTreeID, rtBVersion.RevTreeID)
+	}, time.Second*5, time.Millisecond*100)
+
+	// wait for XDCR stats to ensure attachment data also made it over
+	require.EventuallyWithT(t, func(c *assert.CollectT) {
+		stats, err := xdcrAtoB.Stats(ctx)
+		require.NoError(c, err)
+		// doc and attachment doc
+		assert.Equalf(c, uint64(2), stats.DocsWritten, "expected doc and attachment to be replicated")
+	}, time.Second*5, time.Millisecond*100)
+
+	// stop replication
+	require.NoError(t, xdcrAtoB.Stop(ctx))
+	require.NoError(t, xdcrBtoA.Stop(ctx))
+
+	// fetch attachment via REST API
+	attBResp := rtB.SendAdminRequest(http.MethodGet, "/{{.keyspace}}/"+docID+"/"+attachmentID, "")
+	rest.RequireStatus(t, attBResp, http.StatusOK)
+
+	// update doc on A, removing attachment
+	rtAVersion = rtA.UpdateDoc(docID, rtAVersion, `{"key":"value2"}`)
+
+	// update doc on B, changing body but keeping attachment stub (twice to ensure MWW resolves this as the winner as well as LWW)
+	rtBVersion = rtB.UpdateDoc(docID, rtBVersion, `{"key":"value3","_attachments":{"`+attachmentID+`":{"stub":true}}}`)
+	rtBVersion = rtB.UpdateDoc(docID, rtBVersion, `{"key":"value4","_attachments":{"`+attachmentID+`":{"stub":true}}}`)
+
+	// start replication
+	require.NoError(t, xdcrAtoB.Start(ctx))
+	require.NoError(t, xdcrBtoA.Start(ctx))
+
+	// wait for XDCR stats to ensure attachment data also made it over
+	var expectedDocsWritten uint64 = 0 // attachment deletion (or lack of)
+	// Rosmar's XDCR implementation differs in two ways:
+	// 1. Stats don't get reset on restart
+	// 2. No DCP checkpointing - so there's always more TargetNewerDocs than expected even if we reset stats
+	if !base.TestUseCouchbaseServer() {
+		expectedDocsWritten = expectedDocsWritten + 2 // (old replication stats: 1 doc + 1 attachment)
+	}
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		stats, err := xdcrAtoB.Stats(ctx)
+		require.NoError(c, err)
+		assert.Equalf(c, expectedDocsWritten, stats.DocsWritten, "unexpected additional mutation replicated (an attachment delete?)")
+	}, time.Second*5, time.Millisecond*100)
+
+	// wait for XDCR stats to ensure attachment data also made it over
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		stats, err := xdcrBtoA.Stats(ctx)
+		require.NoError(c, err)
+		assert.Equalf(c, uint64(1), stats.DocsWritten, "expected resolved conflict to be replicated back to A")
+	}, time.Second*5, time.Millisecond*100)
+
+	// wait for doc (resolved conflict) to replicate back to rtA
+	assert.EventuallyWithT(t, func(c *assert.CollectT) {
+		currentRtAVersion, _ := rtA.GetDoc(docID)
+		assert.Equal(c, rtBVersion.CV.String(), currentRtAVersion.CV.String())
+	}, time.Second*10, time.Millisecond*100)
+
+	// check attachment metadata exists
+	docA := rtA.GetDocument(docID)
+	docB := rtB.GetDocument(docID)
+	assert.Equal(t, docA.Attachments(), docB.Attachments())
+	assert.Contains(t, docA.Attachments(), attachmentID)
+	assert.Contains(t, docB.Attachments(), attachmentID)
+
+	// check attachment contents are retrievable
+	attAResp = rtA.SendAdminRequest(http.MethodGet, "/{{.keyspace}}/"+docID+"/"+attachmentID, "")
+	rest.AssertStatus(t, attAResp, http.StatusOK)
+	attBResp = rtB.SendAdminRequest(http.MethodGet, "/{{.keyspace}}/"+docID+"/"+attachmentID, "")
+	rest.AssertStatus(t, attBResp, http.StatusOK)
+
+	require.NoError(t, xdcrAtoB.Stop(ctx))
+	require.NoError(t, xdcrBtoA.Stop(ctx))
+}

xdcr/rosmar_xdcr.go

@@ -178,7 +178,7 @@ func (r *rosmarManager) processEvent(ctx context.Context, event sgbucket.FeedEve
 		*/
 
 		if conflictResolutionSourceCas <= conflictResolutionTargetCas {
-			base.InfofCtx(ctx, base.KeyVV, "XDCR doc:%s skipping replication since sourceCas (%d) < targetCas (%d)", docID, conflictResolutionSourceCas, conflictResolutionTargetCas)
+			base.InfofCtx(ctx, base.KeyVV, "XDCR doc:%s skipping replication since sourceCas (%d) <= targetCas (%d)", docID, conflictResolutionSourceCas, conflictResolutionTargetCas)
 			r.targetNewerDocs.Add(1)
 			return true
 		} /* else if sourceCas == targetCas {