Merge remote-tracking branch 'origin/main' into CBG-5509

Author: torcolvin

.github/workflows/ci.yml

@@ -35,7 +35,7 @@ env:
   ACTIONS_CHECKOUT_VERSION: &actions_checkout_version actions/checkout@v6
   ACTIONS_SETUP_GO_VERSION: &actions_setup_go_version actions/setup-go@v6
   ACTIONS_RUFF_VERSION: &actions_ruff_version astral-sh/ruff-action@v4.0.0
-  GO_VERSION: &go_version 1.26.4
+  GO_VERSION: &go_version 1.26.5
   GOTESTSUM_VERSION: 1.13.0
   GO_TEST_ANNOTATIONS_VERSION: &go_test_annotations_version guyarb/golang-test-annotations@v0.9.0
 

.github/workflows/service.yml

@@ -47,7 +47,7 @@ jobs:
       # build sync gateway since the executable is needed for service installation
       - uses: actions/setup-go@v6
         with:
-          go-version: 1.26.4
+          go-version: 1.26.5
       - name: "Build Sync Gateway"
         run: mkdir -p ./bin && go build -o ./bin ./...
       - name: "Run test scripts"

base/collection_rangescan.go

@@ -84,10 +84,28 @@ func (it *gocbScanResultIterator) Next(_ context.Context) *sgbucket.ScanResultIt
 	return result
 }
 
+func (it *gocbScanResultIterator) Err() error {
+	if it.err != nil {
+		return it.err
+	}
+	return it.result.Err()
+}
+
 func (it *gocbScanResultIterator) Close(_ context.Context) error {
-	closeErr := it.result.Close()
-	if it.err == nil {
+	if it.err != nil {
+		// A real error is already recorded (e.g. a content-decode failure);
+		// release the stream but keep the first error.
+		_ = it.result.Close()
+		return it.err
+	}
+	// gocb's Close returns the first recorded stream error, or nil after
+	// cancelling a clean scan.
+	if closeErr := it.result.Close(); closeErr != nil {
 		it.err = closeErr
+		return it.err
 	}
-	return it.err
+	// Clean end-of-stream: record ErrScanCancelled so a later Err reports it
+	// (matching Rosmar) rather than leaking gocb's internal ErrRequestCanceled.
+	it.err = sgbucket.ErrScanCancelled
+	return nil
 }

base/collection_rangescan_test.go

@@ -33,6 +33,8 @@ func collectScanIDs(t testing.TB, ctx context.Context, rss sgbucket.RangeScanSto
 	for item := iter.Next(ctx); item != nil; item = iter.Next(ctx) {
 		ids = append(ids, item.ID)
 	}
+	// A clean drain must leave no error, inspectable without closing the iterator.
+	assert.NoError(t, iter.Err())
 	sort.Strings(ids)
 	return ids
 }
@@ -82,6 +84,7 @@ func runRangeScanSubtests(t *testing.T, ctx context.Context, writeDS sgbucket.Da
 			assert.NotNil(t, item.Body, "Expected body for key %s", item.ID)
 			assert.NotZero(t, item.Cas, "Expected non-zero CAS for key %s", item.ID)
 		}
+		assert.NoError(t, iter.Err())
 		sort.Strings(ids)
 		require.Equal(t, allDocIDs, ids)
 	})
@@ -105,6 +108,7 @@ func runRangeScanSubtests(t *testing.T, ctx context.Context, writeDS sgbucket.Da
 			ids = append(ids, item.ID)
 			assert.Nil(t, item.Body, "Expected nil body for IDsOnly scan, key %s", item.ID)
 		}
+		assert.NoError(t, iter.Err())
 		sort.Strings(ids)
 		require.Equal(t, allDocIDs, ids)
 	})
@@ -114,6 +118,21 @@ func runRangeScanSubtests(t *testing.T, ctx context.Context, writeDS sgbucket.Da
 		assert.Empty(t, ids)
 	})
 
+	t.Run("CloseCancellation", func(t *testing.T) {
+		iter, err := scanStore.Scan(ctx, sgbucket.NewRangeScanForPrefix("doc_"), sgbucket.ScanOptions{IDsOnly: true})
+		require.NoError(t, err)
+		for item := iter.Next(ctx); item != nil; item = iter.Next(ctx) {
+			require.NotEmpty(t, item.ID)
+		}
+		// Clean end-of-stream: no error is reported until Close is called.
+		require.NoError(t, iter.Err())
+		// A clean Close returns nil...
+		require.NoError(t, iter.Close(ctx))
+		// ...but records a cancellation, surfaced by a later Err (gocb parity),
+		// identically on the Rosmar and gocb-backed implementations.
+		require.ErrorIs(t, iter.Err(), sgbucket.ErrScanCancelled)
+	})
+
 	t.Run("PrefixScan", func(t *testing.T) {
 		ids := collectScanIDs(t, ctx, scanStore, sgbucket.NewRangeScanForPrefix("doc_c"), sgbucket.ScanOptions{})
 		require.Equal(t, []string{"doc_c"}, ids)

base/dcp_client_test.go

@@ -937,8 +937,7 @@ func TestDCPCheckpointCleanup(t *testing.T) {
 	foundDocs := make(chan string, len(dataStores))
 	callback := func(event sgbucket.FeedEvent) bool {
 		if strings.HasSuffix(string(event.Key), "_doc") {
-			mutationCount.Add(1)
-			if mutationCount.Load() == uint64(len(dataStores)) {
+			if mutationCount.Add(1) == uint64(len(dataStores)) {
 				close(foundDocs)
 			}
 		}

base/stats.go

@@ -1263,12 +1263,24 @@ func (s *SgwFloatStat) Add(delta float64) {
 	}
 }
 
+// formatValue renders the stat's current value as a JSON-safe string. JSON has no representation for
+// non-finite floats (+Inf/-Inf/NaN); emitting them as raw bytes ("+Inf" etc.) produces invalid JSON
+// that aborts the entire stats marshal (and the expvar output), so a single degenerate stat could
+// otherwise corrupt the whole serialized blob. Fall back to "0" in that case.
+func (s *SgwFloatStat) formatValue() string {
+	val := math.Float64frombits(atomic.LoadUint64(&s.Val))
+	if math.IsInf(val, 0) || math.IsNaN(val) {
+		return "0"
+	}
+	return strconv.FormatFloat(val, 'g', -1, 64)
+}
+
 func (s *SgwFloatStat) MarshalJSON() ([]byte, error) {
-	return []byte(strconv.FormatFloat(math.Float64frombits(atomic.LoadUint64(&s.Val)), 'g', -1, 64)), nil
+	return []byte(s.formatValue()), nil
 }
 
 func (s *SgwFloatStat) String() string {
-	return strconv.FormatFloat(math.Float64frombits(atomic.LoadUint64(&s.Val)), 'g', -1, 64)
+	return s.formatValue()
 }
 
 func (s *SgwFloatStat) Value() float64 {

base/stats_test.go

@@ -12,6 +12,7 @@ package base
 
 import (
 	"expvar"
+	"math"
 	"testing"
 
 	"github.com/couchbase/sync_gateway/testing/require"
@@ -162,3 +163,39 @@ func initExpvarBaseEquivalent() *expvar.Map {
 
 	return expvarMap
 }
+
+// TestSgwFloatStatMarshalNonFinite is a regression test for CBG-3658: a non-finite float value
+// (e.g. +Inf produced by a divide-by-zero in the process CPU percentage calculation) must not
+// corrupt stats serialization. Previously MarshalJSON emitted the raw bytes "+Inf"/"NaN", which the
+// JSON encoder rejected ("invalid character '+' looking for beginning of value"), collapsing the
+// entire SgwStats blob to "null".
+func TestSgwFloatStatMarshalNonFinite(t *testing.T) {
+	for _, test := range []struct {
+		name  string
+		value float64
+	}{
+		{"positive_infinity", math.Inf(1)},
+		{"negative_infinity", math.Inf(-1)},
+		{"nan", math.NaN()},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			stat := &SgwFloatStat{}
+			stat.Set(test.value)
+
+			// Marshal the stat inside a larger structure: this is the exact path that failed via
+			// SgwStats.String -> JSONMarshalCanonical.
+			marshalled, err := JSONMarshalCanonical(map[string]*SgwFloatStat{"stat": stat})
+			require.NoError(t, err)
+
+			// The output must be valid, parseable JSON, and non-finite values must fall back to 0.
+			var roundTripped map[string]float64
+			require.NoError(t, JSONUnmarshal(marshalled, &roundTripped))
+			require.Equal(t, 0.0, roundTripped["stat"])
+
+			// String() satisfies expvar.Var, which likewise requires a valid JSON value.
+			var viaString float64
+			require.NoError(t, JSONUnmarshal([]byte(stat.String()), &viaString))
+			require.Equal(t, 0.0, viaString)
+		})
+	}
+}

base/util_testing.go

@@ -794,12 +794,12 @@ func AssertWaitForStat(t testing.TB, getStatFunc func() int64, expected int64) (
 }
 
 // RequireWaitForStat will retry for up to 20 seconds until the result of getStatFunc is equal to the expected value.
-func RequireWaitForStat(t testing.TB, getStatFunc func() int64, expected int64) (val int64) {
+func RequireWaitForStat(t testing.TB, getStatFunc func() int64, expected int64, msgAndArgs ...any) (val int64) {
 	t.Helper()
 	require.NotNil(t, getStatFunc, "Function for RequireWaitForStat cannot be nil")
 	require.EventuallyWithT(t, func(c *assert.CollectT) {
 		val = getStatFunc()
-		assert.Equal(c, expected, val)
+		assert.Equal(c, expected, val, msgAndArgs...)
 	}, 20*time.Second, 100*time.Millisecond)
 	return val
 }

db/attachment_test.go

@@ -19,6 +19,7 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 
 	"github.com/couchbase/sync_gateway/base"
@@ -1521,3 +1522,150 @@ func TestLargeAttachments(t *testing.T) {
 	require.ErrorAs(t, err, &httpErr, "Created doc with huge attachment")
 	require.Equal(t, http.StatusRequestEntityTooLarge, httpErr.Status)
 }
+
+// attachmentMigrationCASRetrySetup creates a doc with an attachment, then moves the attachment
+// back to AttachmentsPre4dot0 to simulate a document that has not yet been through the background
+// migration.
+func attachmentMigrationCASRetrySetup(t *testing.T, ctx context.Context, collection *DatabaseCollectionWithUser, docID string) {
+	t.Helper()
+	_, _, err := collection.Put(ctx, docID, Body{
+		"test": "doc",
+		BodyAttachments: AttachmentsMeta{
+			"camera.txt": map[string]any{"data": "Q2Fub24gRU9TIDVEIE1hcmsgSVY="},
+		},
+	})
+	require.NoError(t, err)
+
+	ds := collection.GetCollectionDatastore()
+	xattrs, cas, err := ds.GetXattrs(ctx, docID, []string{base.SyncXattrName, base.GlobalXattrName})
+	require.NoError(t, err)
+	var bucketSyncData SyncData
+	require.NoError(t, base.JSONUnmarshal(xattrs[base.SyncXattrName], &bucketSyncData))
+	var globalXattr GlobalSyncData
+	require.NoError(t, base.JSONUnmarshal(xattrs[base.GlobalXattrName], &globalXattr))
+
+	bucketSyncData.AttachmentsPre4dot0 = globalXattr.Attachments
+	syncBytes := base.MustJSONMarshal(t, bucketSyncData)
+	_, err = ds.WriteWithXattrs(ctx, docID, 0, cas, []byte(`{"test":"doc"}`),
+		map[string][]byte{base.SyncXattrName: syncBytes}, []string{base.GlobalXattrName}, nil)
+	require.NoError(t, err)
+}
+
+// attachmentMigrationCASRetryCallback returns an UpdateCallback that calls MigrateAttachmentMetadata
+// exactly once for the given docID, changing the document CAS and forcing WriteUpdateWithXattrs to retry.
+func attachmentMigrationCASRetryCallback(t *testing.T, ctx context.Context, collection *DatabaseCollectionWithUser, docID string) func(string) {
+	t.Helper()
+	var once sync.Once
+	return func(key string) {
+		if key != docID {
+			return
+		}
+		once.Do(func() {
+			ds := collection.GetCollectionDatastore()
+			xattrs, cas, err := ds.GetXattrs(ctx, docID, []string{base.SyncXattrName})
+			require.NoError(t, err)
+			var syncData SyncData
+			require.NoError(t, base.JSONUnmarshal(xattrs[base.SyncXattrName], &syncData))
+			require.NoError(t, collection.MigrateAttachmentMetadata(ctx, docID, cas, &syncData))
+		})
+	}
+}
+
+const attachmentMigrationCASRetryDigest = "sha1-VoSNiNQGHE1HirIS5HMxj6CrlHI="
+const attachmentMigrationCASRetryData = "Q2Fub24gRU9TIDVEIE1hcmsgSVY="
+
+var attachmentMigrationCASRetryExpected = AttachmentMap{
+	"camera.txt": {
+		Digest:  attachmentMigrationCASRetryDigest,
+		Length:  20,
+		Revpos:  2,
+		Version: 2,
+		Stub:    true,
+	},
+}
+
+// TestAttachmentMigrationCASRetryOnUpdate verifies that PutExistingCurrentVersion and
+// PutExistingRevWithConflictResolution both preserve revpos=2 when a CAS retry is forced
+// by concurrent attachment migration. The maps.Clone reset in each function is the guard:
+// without it, the retry sees a stub mutated by the first invocation and applies a parent
+// lookup that returns the migrated revpos=1.
+func TestAttachmentMigrationCASRetryOnUpdate(t *testing.T) {
+	db, ctx := setupTestLeakyDBWithCacheOptions(t, DefaultCacheOptions(), base.LeakyBucketConfig{})
+	defer db.Close(ctx)
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	ds := collection.GetCollectionDatastore()
+
+	leakyDS, ok := base.AsLeakyDataStore(ds)
+	require.True(t, ok)
+
+	// putCurrentVersion calls PutExistingCurrentVersion (HLV/versionVector BLIP path).
+	putCurrentVersion := func(t *testing.T, docID string) {
+		existingDoc, err := collection.GetDocument(ctx, docID, DocUnmarshalSync)
+		require.NoError(t, err)
+		incomingHLV := &HybridLogicalVector{
+			SourceID:         "cbl1",
+			Version:          existingDoc.HLV.Version + 1,
+			PreviousVersions: HLVVersions{existingDoc.HLV.SourceID: existingDoc.HLV.Version},
+		}
+		// newDoc carries the attachment as inline data, simulating what ForEachStubAttachment
+		// places in newDoc.Attachments() after downloading the data from the CBL client.
+		newDoc := &Document{ID: docID}
+		newDoc.SetAttachments(AttachmentsMeta{
+			"camera.txt": map[string]any{"data": attachmentMigrationCASRetryData},
+		})
+		newDoc.UpdateBody(Body{"some": "update"})
+		_, _, _, err = collection.PutExistingCurrentVersion(ctx, PutDocOptions{
+			NewDoc:    newDoc,
+			NewDocHLV: incomingHLV,
+		})
+		require.NoError(t, err)
+	}
+
+	// putExistingRev calls PutExistingRevWithConflictResolution (rev-tree BLIP path).
+	putExistingRev := func(t *testing.T, docID string) {
+		existingDoc, err := collection.GetDocument(ctx, docID, DocUnmarshalSync)
+		require.NoError(t, err)
+		existingRevID := existingDoc.GetRevTreeID()
+		incomingRevID := "2-attachmentmigrationcasretrytest"
+		newDoc := &Document{ID: docID, RevID: incomingRevID}
+		newDoc.SetAttachments(AttachmentsMeta{
+			"camera.txt": map[string]any{"data": attachmentMigrationCASRetryData},
+		})
+		newDoc.UpdateBody(Body{"some": "update"})
+		_, _, err = collection.PutExistingRevWithConflictResolution(ctx, PutDocOptions{
+			NewDoc:         newDoc,
+			RevTreeHistory: []string{incomingRevID, existingRevID},
+			DocUpdateEvent: ExistingVersionWithUpdateToHLV,
+		})
+		require.NoError(t, err)
+	}
+
+	testCases := []struct {
+		name       string
+		casFailure bool
+		update     func(t *testing.T, docID string)
+	}{
+		{name: "PutExistingCurrentVersion no CAS failure", casFailure: false, update: putCurrentVersion},
+		{name: "PutExistingCurrentVersion CAS failure", casFailure: true, update: putCurrentVersion},
+		{name: "PutExistingRevWithConflictResolution no CAS failure", casFailure: false, update: putExistingRev},
+		{name: "PutExistingRevWithConflictResolution CAS failure", casFailure: true, update: putExistingRev},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			docID := "doc_" + strings.ReplaceAll(tc.name, " ", "_")
+
+			attachmentMigrationCASRetrySetup(t, ctx, collection, docID)
+
+			if tc.casFailure {
+				leakyDS.SetUpdateCallback(attachmentMigrationCASRetryCallback(t, ctx, collection, docID))
+				defer leakyDS.SetUpdateCallback(nil)
+			}
+
+			tc.update(t, docID)
+
+			require.Empty(t, GetRawSyncXattr(t, ds, docID).AttachmentsPre4dot0)
+			require.Equal(t, attachmentMigrationCASRetryExpected, GetRawGlobalSyncAttachments(t, ds, docID))
+		})
+	}
+}