CBG-4578: have maximum threshold for releasing sequences in nextSequenceGreaterThan (#7436)

Author: gregns1

base/error.go

@@ -77,6 +77,9 @@ var (
 
 	// ErrSkippedSequencesMissing is returned when attempting to remove a sequence range form the skipped sequence list and at least one sequence in that range is not present
 	ErrSkippedSequencesMissing = &sgError{"Sequence range has sequences that aren't present in skipped list"}
+
+	// ErrMaxSequenceReleasedExceeded is returned when the maximum number of sequences to be released as part of nextSequenceGreaterThan is exceeded
+	ErrMaxSequenceReleasedExceeded = &sgError{"Maximum number of sequences to release to catch up with document sequence exceeded"}
 )
 
 func (e *sgError) Error() string {

base/stats.go

@@ -645,6 +645,8 @@ type DatabaseStats struct {
 	SequenceReleasedCount *SgwIntStat `json:"sequence_released_count"`
 	// The total number of sequences reserved by Sync Gateway.
 	SequenceReservedCount *SgwIntStat `json:"sequence_reserved_count"`
+	// The total number of corrupt sequences above the MaxSequencesToRelease threshold seen at the sequence allocator
+	CorruptSequenceCount *SgwIntStat `json:"corrupt_sequence_count"`
 	// The total number of warnings relating to the channel name size.
 	WarnChannelNameSizeCount *SgwIntStat `json:"warn_channel_name_size_count"`
 	// The total number of warnings relating to the channel count exceeding the channel count threshold.
@@ -1754,6 +1756,10 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
+	resUtil.CorruptSequenceCount, err = NewIntStat(SubsystemDatabaseKey, "corrupt_sequence_count", StatUnitNoUnits, CorruptSequenceCountDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	if err != nil {
+		return err
+	}
 	resUtil.WarnChannelNameSizeCount, err = NewIntStat(SubsystemDatabaseKey, "warn_channel_name_size_count", StatUnitNoUnits, WarnChannelNameSizeCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
@@ -1839,6 +1845,7 @@ func (d *DbStats) unregisterDatabaseStats() {
 	prometheus.Unregister(d.DatabaseStats.SequenceIncrCount)
 	prometheus.Unregister(d.DatabaseStats.SequenceReleasedCount)
 	prometheus.Unregister(d.DatabaseStats.SequenceReservedCount)
+	prometheus.Unregister(d.DatabaseStats.CorruptSequenceCount)
 	prometheus.Unregister(d.DatabaseStats.WarnChannelNameSizeCount)
 	prometheus.Unregister(d.DatabaseStats.WarnChannelsPerDocCount)
 	prometheus.Unregister(d.DatabaseStats.WarnGrantsPerDocCount)

base/stats_descriptions.go

@@ -320,6 +320,9 @@ const (
 
 	TotalInitFatalErrorsDesc   = "The total number of errors that occurred that prevented the database from being initialized."
 	TotalOnlineFatalErrorsDesc = "The total number of errors that occurred that prevented the database from being brought online."
+
+	CorruptSequenceCountDesc = "The total number of corrupt sequences detected at the sequence allocator. Documents that have a corrupt " +
+		"sequence that trigger release of sequences above the MaxSequenceToRelease threshold will have their update cancelled."
 )
 
 // Delta Sync stats descriptions

db/crud.go

@@ -1882,6 +1882,9 @@ func (col *DatabaseCollectionWithUser) documentUpdateFunc(ctx context.Context, d
 
 	unusedSequences, err = col.assignSequence(ctx, previousDocSequenceIn, doc, unusedSequences)
 	if err != nil {
+		if errors.Is(err, base.ErrMaxSequenceReleasedExceeded) {
+			base.ErrorfCtx(ctx, "Doc %s / %s had a much larger sequence (%d) than the current sequence number. Document update will be cancelled, since we don't want to allocate sequences to fill a gap this large. This may indicate document metadata being migrated between databases where it should've been stripped and re-imported.", base.UD(newDoc.ID), prevCurrentRev, doc.Sequence)
+		}
 		return
 	}
 

db/crud_test.go

@@ -1761,3 +1761,46 @@ func TestReleaseSequenceOnDocWriteFailure(t *testing.T) {
 		assert.Equal(t, int64(1), db.DbStats.Database().SequenceReleasedCount.Value())
 	}, time.Second*10, time.Millisecond*100)
 }
+
+func TestDocUpdateCorruptSequence(t *testing.T) {
+	if !base.TestUseXattrs() {
+		t.Skip("This test only works with XATTRS enabled")
+	}
+	base.SetUpTestLogging(t, base.LevelInfo, base.KeyCache, base.KeyChanges, base.KeyCRUD, base.KeyDCP)
+
+	db, ctx := SetupTestDBWithOptions(t, DatabaseContextOptions{})
+	defer db.Close(ctx)
+
+	// create a sequence much higher than _syc:seqs value
+	const corruptSequence = MaxSequencesToRelease + 1000
+
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+	rev, doc, err := collection.Put(ctx, "doc1", Body{"foo": "bar"})
+	require.NoError(t, err)
+	docRev := doc.RevID
+	t.Logf("doc sequence: %d", doc.Sequence)
+
+	// but we can fiddle with the sequence in the metadata of the doc write to simulate a doc from a different cluster (with a higher sequence)
+	_, xattrs, _, err := collection.dataStore.GetWithXattrs(ctx, "doc1", []string{base.SyncXattrName})
+	require.NoError(t, err)
+	var newSyncData map[string]interface{}
+	err = json.Unmarshal(xattrs[base.SyncXattrName], &newSyncData)
+	require.NoError(t, err)
+	newSyncData["sequence"] = corruptSequence
+	_, err = collection.dataStore.UpdateXattrs(ctx, doc.ID, 0, doc.Cas, map[string][]byte{base.SyncXattrName: base.MustJSONMarshal(t, newSyncData)}, DefaultMutateInOpts())
+	require.NoError(t, err)
+
+	_, _, err = collection.Put(ctx, "doc1", Body{"foo": "buzz", BodyRev: rev})
+	require.Error(t, err)
+	require.ErrorIs(t, err, base.ErrMaxSequenceReleasedExceeded)
+
+	// assert update to doc was cancelled thus doc1 is its original version
+	doc, err = collection.GetDocument(ctx, "doc1", DocUnmarshalAll)
+	require.NoError(t, err)
+	assert.Equal(t, uint64(corruptSequence), doc.Sequence)
+	assert.Equal(t, docRev, doc.RevID)
+
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.Database().CorruptSequenceCount.Value()
+	}, 1)
+}

db/import_test.go

@@ -12,8 +12,10 @@ package db
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"log"
+	"net/http"
 	"testing"
 	"time"
 
@@ -1019,3 +1021,99 @@ func getSyncAndMou(t *testing.T, collection *DatabaseCollectionWithUser, key str
 	return syncData, mou, cas
 
 }
+
+func TestImportCancelOnDocWithCorruptSequenceOverImportFeed(t *testing.T) {
+	if !base.TestUseXattrs() {
+		t.Skip("This test only works with XATTRS enabled")
+	}
+
+	base.SetUpTestLogging(t, base.LevelInfo, base.KeyImport, base.KeyCRUD)
+	db, ctx := setupTestDBWithOptionsAndImport(t, nil, DatabaseContextOptions{})
+	defer db.Close(ctx)
+
+	// create a sequence much higher than _syc:seqs value
+	const corruptSequence = MaxSequencesToRelease + 1000
+
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+
+	key := t.Name()
+	bodyBytes := []byte(`{"foo":"bar"}`)
+	// Create via the SDK
+	_, err := collection.dataStore.AddRaw(key, 0, bodyBytes)
+	require.NoError(t, err)
+
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.SharedBucketImport().ImportCount.Value()
+	}, 1)
+
+	_, xattrs, cas, err := collection.dataStore.GetWithXattrs(ctx, key, []string{base.SyncXattrName})
+	require.NoError(t, err)
+
+	// corrupt the document sequence
+	var newSyncData map[string]interface{}
+	err = json.Unmarshal(xattrs[base.SyncXattrName], &newSyncData)
+	require.NoError(t, err)
+	newSyncData["sequence"] = corruptSequence
+	_, err = collection.dataStore.UpdateXattrs(ctx, key, 0, cas, map[string][]byte{base.SyncXattrName: base.MustJSONMarshal(t, newSyncData)}, DefaultMutateInOpts())
+	require.NoError(t, err)
+
+	// sdk update to trigger import
+	require.NoError(t, collection.dataStore.SetRaw(key, 0, nil, []byte(`{"foo":"baz"}`)))
+
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.SharedBucketImport().ImportErrorCount.Value()
+	}, 1)
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.Database().CorruptSequenceCount.Value()
+	}, 1)
+}
+
+func TestImportCancelOnDocWithCorruptSequenceOndemand(t *testing.T) {
+	if !base.TestUseXattrs() {
+		t.Skip("This test only works with XATTRS enabled")
+	}
+
+	base.SetUpTestLogging(t, base.LevelInfo, base.KeyImport, base.KeyCRUD)
+	tb := base.GetTestBucket(t)
+	defer tb.Close(base.TestCtx(t))
+	db, ctx := SetupTestDBForDataStoreWithOptions(t, tb, DatabaseContextOptions{})
+	key := t.Name()
+
+	collection, ctx := GetSingleDatabaseCollectionWithUser(ctx, t, db)
+
+	_, _, err := collection.Put(ctx, key, Body{"foo": "bar"})
+	require.NoError(t, err)
+
+	// create a sequence much higher than _syc:seqs value
+	const corruptSequence = MaxSequencesToRelease + 1000
+
+	_, xattrs, cas, err := collection.dataStore.GetWithXattrs(ctx, key, []string{base.SyncXattrName})
+	require.NoError(t, err)
+
+	// corrupt the document sequence
+	var newSyncData map[string]interface{}
+	err = json.Unmarshal(xattrs[base.SyncXattrName], &newSyncData)
+	require.NoError(t, err)
+	newSyncData["sequence"] = corruptSequence
+	_, err = collection.dataStore.UpdateXattrs(ctx, key, 0, cas, map[string][]byte{base.SyncXattrName: base.MustJSONMarshal(t, newSyncData)}, DefaultMutateInOpts())
+	require.NoError(t, err)
+
+	// sdk update
+	require.NoError(t, collection.dataStore.SetRaw(key, 0, nil, []byte(`{"foo":"baz"}`)))
+
+	// trigger on demand import
+	_, err = collection.GetDocument(ctx, key, DocUnmarshalAll)
+	require.Error(t, err)
+	var httpErr *base.HTTPError
+	require.ErrorAs(t, err, &httpErr)
+	assert.Equal(t, http.StatusNotFound, httpErr.Status)
+
+	// verify that the document was not imported
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.SharedBucketImport().ImportErrorCount.Value()
+	}, 1)
+	base.RequireWaitForStat(t, func() int64 {
+		return db.DbStats.Database().CorruptSequenceCount.Value()
+	}, 1)
+
+}

db/sequence_allocator.go

@@ -39,6 +39,9 @@ const (
 	// Accounts for a maximum SG cluster size of 50 nodes, each allocating a full batch size of 10
 	// if this value is too low and this correction has potential to allocate sequences that other nodes have already reserved a batch for
 	syncSeqCorrectionValue = 500
+
+	// Maximum number of sequences to release as part of nextSequenceGreaterThan
+	MaxSequencesToRelease = 10000000
 )
 
 // MaxSequenceIncrFrequency is the maximum frequency we want to perform incr operations.
@@ -286,6 +289,14 @@ func (s *sequenceAllocator) nextSequenceGreaterThan(ctx context.Context, existin
 	numberToRelease := existingSequence - syncSeq
 	numberToAllocate := s.sequenceBatchSize
 	incrVal := numberToRelease + numberToAllocate
+
+	// if sequences to release are above the max allowed, return error to cancel update
+	if numberToRelease > MaxSequencesToRelease {
+		s.mutex.Unlock()
+		s.dbStats.CorruptSequenceCount.Add(1) // increment corrupt sequence count
+		return 0, 0, base.ErrMaxSequenceReleasedExceeded
+	}
+
 	allocatedToSeq, err := s.incrementSequence(incrVal)
 	if err != nil {
 		base.WarnfCtx(ctx, "Error from incrementSequence in nextSequenceGreaterThan(%d): %v", existingSequence, err)

rest/adminapitest/resync_test.go

@@ -9,6 +9,7 @@
 package adminapitest
 
 import (
+	"encoding/json"
 	"fmt"
 	"net/http"
 	"testing"
@@ -69,6 +70,64 @@ func TestResyncRollback(t *testing.T) {
 	status = rt.WaitForResyncDCPStatus(db.BackgroundProcessStateCompleted)
 }
 
+func TestResyncRegenerateSequencesCorruptDocumentSequence(t *testing.T) {
+	if base.UnitTestUrlIsWalrus() {
+		t.Skip("This test doesn't works with walrus")
+	}
+	base.SetUpTestLogging(t, base.LevelInfo, base.KeyCRUD, base.KeyChanges, base.KeyAccess)
+	rt := rest.NewRestTester(t, &rest.RestTesterConfig{
+		AutoImport: base.BoolPtr(false),
+	})
+
+	defer rt.Close()
+	ctx := base.TestCtx(t)
+	ds := rt.GetSingleDataStore()
+
+	// create a sequence much higher than _syc:seqs value
+	const corruptSequence = db.MaxSequencesToRelease + 1000
+
+	numDocs := 10
+	for i := 0; i < numDocs; i++ {
+		rt.CreateTestDoc(fmt.Sprintf("doc%v", i))
+	}
+
+	// corrupt doc0 sequence
+	_, xattrs, cas, err := ds.GetWithXattrs(ctx, "doc0", []string{base.SyncXattrName})
+	require.NoError(t, err)
+
+	// corrupt the document sequence
+	var newSyncData map[string]interface{}
+	err = json.Unmarshal(xattrs[base.SyncXattrName], &newSyncData)
+	require.NoError(t, err)
+	newSyncData["sequence"] = corruptSequence
+	_, err = ds.UpdateXattrs(ctx, "doc0", 0, cas, map[string][]byte{base.SyncXattrName: base.MustJSONMarshal(t, newSyncData)}, nil)
+	require.NoError(t, err)
+
+	response := rt.SendAdminRequest("POST", "/{{.db}}/_offline", "")
+	rest.RequireStatus(t, response, http.StatusOK)
+	rt.WaitForDBState(db.RunStateString[db.DBOffline])
+
+	// we need to wait for the resync to start and not finish
+	resp := rt.SendAdminRequest("POST", "/{{.db}}/_resync?action=start&regenerate_sequences=true", "")
+	rest.RequireStatus(t, resp, http.StatusOK)
+	_ = rt.WaitForResyncDCPStatus(db.BackgroundProcessStateRunning)
+
+	_ = rt.WaitForResyncDCPStatus(db.BackgroundProcessStateCompleted)
+
+	_, xattrs, _, err = ds.GetWithXattrs(ctx, "doc0", []string{base.SyncXattrName})
+	require.NoError(t, err)
+	// assert doc sequence wasn't changed
+	var bucketSync map[string]interface{}
+	err = json.Unmarshal(xattrs[base.SyncXattrName], &bucketSync)
+	require.NoError(t, err)
+	seq := newSyncData["sequence"].(int)
+	require.Equal(t, uint64(corruptSequence), uint64(seq))
+
+	base.RequireWaitForStat(t, func() int64 {
+		return rt.GetDatabase().DbStats.Database().CorruptSequenceCount.Value()
+	}, 1)
+}
+
 func TestResyncRegenerateSequencesPrincipals(t *testing.T) {
 	base.TestRequiresDCPResync(t)
 	if !base.TestsUseNamedCollections() {