Merge remote-tracking branch 'origin/main' into CBG-4345-thread-err

Author: torcolvin

auth/auth_test.go

@@ -493,7 +493,6 @@ func TestRebuildChannelsError(t *testing.T) {
 	auth := NewTestAuthenticator(t, dataStore, &computer, DefaultAuthenticatorOptions(base.TestCtx(t)))
 	role, err := auth.NewRole("testRole2", ch.BaseSetOf(t, "explicit1"))
 	require.NoError(t, err)
-	assert.NoError(t, err)
 	err = auth.Save(role)
 	require.NoError(t, err)
 

auth/collection_access_test.go

@@ -139,7 +139,8 @@ func TestUserCollectionAccess(t *testing.T) {
 	require.ErrorIs(t, user.authorizeAllChannels(ch.BaseSetOf(t, "*")), errNotAllowedChannels)
 	require.ErrorIs(t, user.authorizeAnyChannel(ch.BaseSetOf(t, "x")), errUnauthorized)
 	require.ErrorIs(t, user.authorizeAnyChannel(ch.BaseSetOf(t, "*")), errUnauthorized)
-	require.ErrorIs(t, user.authorizeAnyChannel(ch.BaseSetOf(t)), errUnauthorized) // Matching named collection checks
+	require.ErrorIs(t, user.authorizeAnyChannel(ch.BaseSetOf(t)), errUnauthorized)
+	// Matching named collection checks
 	requireExpandCollectionWildCardChannels(t, user, scope, collection, []string{"*", "q"}, []string{"*"})
 	requireCanSeeCollectionChannels(t, scope, collection, user, "x", "y", "z", "q", "*")
 	require.NoError(t, user.authorizeAllCollectionChannels(scope, collection, ch.BaseSetOf(t, "x", "y")))

base/bucket.go

@@ -540,21 +540,25 @@ func RequireNoBucketTTL(ctx context.Context, b Bucket) error {
 
 // GetSourceID returns the source ID for a bucket.
 func GetSourceID(ctx context.Context, bucket Bucket) (string, error) {
-	// for rosmar bucket and testing, use the bucket name as the source ID to make it easier to identify the source
+	bucketUUID, err := bucket.UUID()
+	if err != nil {
+		return "", err
+	}
 	gocbBucket, err := AsGocbV2Bucket(bucket)
 	if err != nil {
-		return bucket.GetName(), nil
+		// for rosmar bucket and testing, use the bucket name as the source ID to make it easier to identify the source
+		if underGoTest() {
+			return bucket.GetName(), nil
+		}
+		serverUUID := ""
+		return CreateEncodedSourceID(bucketUUID, serverUUID)
 	}
 
 	// If not overwriting the source ID, for rosmar, serverUUID would be ""
 	serverUUID, err := GetServerUUID(ctx, gocbBucket)
 	if err != nil {
 		return "", err
 	}
-	bucketUUID, err := bucket.UUID()
-	if err != nil {
-		return "", err
-	}
 	return CreateEncodedSourceID(bucketUUID, serverUUID)
 }
 

base/logging_context_test.go

@@ -25,6 +25,9 @@ func requireLogIs(t testing.TB, s string, f func()) {
 
 	config := &ConsoleLoggerConfig{
 		ColorEnabled: Ptr(false),
+		FileLoggerConfig: FileLoggerConfig{
+			Enabled: Ptr(true),
+		},
 	}
 	tempLogger, err := NewConsoleLogger(TestCtx(t), false, config)
 	require.NoError(t, err)
@@ -39,23 +42,11 @@ func requireLogIs(t testing.TB, s string, f func()) {
 
 	f()
 
-	var log string
-	var originalLog string
-	// Allow time for logs to be printed
-	retry := func() (shouldRetry bool, err error, value any) {
-		originalLog = b.String()
-		if len(originalLog) < timestampLength {
-			return false, nil, nil
-		}
-		log = originalLog[timestampLength:]
-		if log == s {
-			return false, nil, nil
-		}
-		return true, nil, nil
-	}
-	err, _ = RetryLoop(TestCtx(t), "wait for logs", retry, CreateSleeperFunc(10, 100))
-
-	require.NoError(t, err, "Console logs did not contain %q, got %q", s, originalLog)
+	FlushLogBuffers()
+	originalLog := b.String()
+	require.GreaterOrEqual(t, len(originalLog), timestampLength, "%q is not a valid log line, needs to contain timestamp", originalLog)
+	log := originalLog[timestampLength:]
+	require.Equal(t, s, log)
 }
 
 func RequireLogMessage(t testing.TB, ctx context.Context, expectedMessage string, logString string) {

base/stats.go

@@ -76,6 +76,7 @@ const (
 	StatUnitUnixTimestamp = "unix timestamp"
 
 	StatFormatInt      = "int"
+	StatFormatUint     = "uint64"
 	StatFormatFloat    = "float"
 	StatFormatDuration = "duration"
 	StatFormatBool     = "bool"
@@ -472,10 +473,10 @@ type CacheStats struct {
 	// The highest sequence number cached.
 	//
 	// There may be skipped sequences lower than high_seq_cached.
-	HighSeqCached *SgwIntStat `json:"high_seq_cached"`
+	HighSeqCached *SgwUint64Stat `json:"high_seq_cached"`
 	// The highest contiguous sequence number that has been cached.
-	HighSeqStable         *SgwIntStat `json:"high_seq_stable"`
-	NonMobileIgnoredCount *SgwIntStat `json:"non_mobile_ignored_count"`
+	HighSeqStable         *SgwUint64Stat `json:"high_seq_stable"`
+	NonMobileIgnoredCount *SgwIntStat    `json:"non_mobile_ignored_count"`
 	// The total number of active channels.
 	NumActiveChannels *SgwIntStat `json:"num_active_channels"`
 	// The total number of skipped sequences. This is a cumulative value.
@@ -606,9 +607,9 @@ type DatabaseStats struct {
 	ReplicationBytesReceived *SgwIntStat `json:"replication_bytes_received"`
 	ReplicationBytesSent     *SgwIntStat `json:"replication_bytes_sent"`
 	// The compaction_attachment_start_time.
-	CompactionAttachmentStartTime *SgwIntStat `json:"compaction_attachment_start_time"`
+	CompactionAttachmentStartTime *SgwUint64Stat `json:"compaction_attachment_start_time"`
 	// The compaction_tombstone_start_time.
-	CompactionTombstoneStartTime *SgwIntStat `json:"compaction_tombstone_start_time"`
+	CompactionTombstoneStartTime *SgwUint64Stat `json:"compaction_tombstone_start_time"`
 	// The total number of writes that left the document in a conflicted state. Includes new conflicts, and mutations that don’t resolve existing conflicts.
 	ConflictWriteCount *SgwIntStat `json:"conflict_write_count"`
 	// The total number of instances during import when the document cas had changed, but the document was not imported because the document body had not changed.
@@ -630,7 +631,7 @@ type DatabaseStats struct {
 	// The total size of xattrs written (in bytes).
 	DocWritesXattrBytes *SgwIntStat `json:"doc_writes_xattr_bytes"`
 	// Highest sequence number seen on the caching DCP feed.
-	HighSeqFeed *SgwIntStat `json:"high_seq_feed"`
+	HighSeqFeed *SgwUint64Stat `json:"high_seq_feed"`
 	// The number of attachments compacted
 	NumAttachmentsCompacted *SgwIntStat `json:"num_attachments_compacted"`
 	// The total number of documents read via Couchbase Lite 2.x replication since Sync Gateway node startup.
@@ -653,19 +654,19 @@ type DatabaseStats struct {
 	// The total amount of bytes read over the public REST api
 	PublicRestBytesRead *SgwIntStat `json:"public_rest_bytes_read"`
 	// The value of the last sequence number assigned. Callers using Set should be holding a mutex or ensure concurrent updates to this value are otherwise safe.
-	LastSequenceAssignedValue *SgwIntStat `json:"last_sequence_assigned_value"` // TODO: CBG-4579 - Replace with SgwUintStat stat
+	LastSequenceAssignedValue *SgwUint64Stat `json:"last_sequence_assigned_value"`
 	// The total number of sequence numbers assigned.
-	SequenceAssignedCount *SgwIntStat `json:"sequence_assigned_count"`
+	SequenceAssignedCount *SgwUint64Stat `json:"sequence_assigned_count"`
 	// The total number of high sequence lookups.
 	SequenceGetCount *SgwIntStat `json:"sequence_get_count"`
 	// The total number of times the sequence counter document has been incremented.
 	SequenceIncrCount *SgwIntStat `json:"sequence_incr_count"`
 	// The total number of unused, reserved sequences released by Sync Gateway.
-	SequenceReleasedCount *SgwIntStat `json:"sequence_released_count"`
+	SequenceReleasedCount *SgwUint64Stat `json:"sequence_released_count"`
 	// The value of the last sequence number reserved (which may not yet be assigned). Callers using Set should be holding a mutex or ensure concurrent updates to this value are otherwise safe.
-	LastSequenceReservedValue *SgwIntStat `json:"last_sequence_reserved_value"` // TODO: CBG-4579 - Replace with SgwUintStat stat
+	LastSequenceReservedValue *SgwUint64Stat `json:"last_sequence_reserved_value"`
 	// The total number of sequences reserved by Sync Gateway.
-	SequenceReservedCount *SgwIntStat `json:"sequence_reserved_count"`
+	SequenceReservedCount *SgwUint64Stat `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.
@@ -986,6 +987,11 @@ type SgwIntStat struct {
 	AtomicInt
 }
 
+type SgwUint64Stat struct {
+	SgwStat
+	atomic.Uint64
+}
+
 // uint64 is used here because atomic ints do not support floats. Floats are encoded to uint64
 type SgwFloatStat struct {
 	SgwStat
@@ -1079,6 +1085,74 @@ func (s *SgwIntStat) String() string {
 	return strconv.FormatInt(s.Value(), 10)
 }
 
+// NewUIntStat creates a uint64 stat. The stat is initialized to initialValue.
+func NewUIntStat(subsystem, key, unit, description, addedVersion, deprecatedVersion, stability string, labelKeys, labelVals []string, statValueType prometheus.ValueType, initialValue uint64) (*SgwUint64Stat, error) {
+	stat, err := newSGWStat(subsystem, key, unit, description, addedVersion, deprecatedVersion, stability, labelKeys, labelVals, statValueType)
+	if err != nil {
+		return nil, err
+	}
+
+	wrappedStat := &SgwUint64Stat{
+		SgwStat: *stat,
+	}
+
+	wrappedStat.Set(initialValue)
+
+	if !SkipPrometheusStatsRegistration {
+		err := prometheus.Register(wrappedStat)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return wrappedStat, nil
+}
+
+// FormatString returns a name of the stat type used for generating documentation.
+func (s *SgwUint64Stat) FormatString() string {
+	return StatFormatUint
+}
+
+func (s *SgwUint64Stat) Describe(ch chan<- *prometheus.Desc) {
+	ch <- s.statDesc
+}
+
+// Collect satisfies the prometheus.Collector interface communicates the stat to prometheus. This is known to truncate to float64, which is the only format that prometheus can accept.
+func (s *SgwUint64Stat) Collect(ch chan<- prometheus.Metric) {
+	ch <- prometheus.MustNewConstMetric(s.statDesc, s.statValueType, float64(s.Value()))
+}
+
+func (s *SgwUint64Stat) MarshalJSON() ([]byte, error) {
+	return []byte(strconv.FormatUint(s.Value(), 10)), nil
+}
+
+func (s *SgwUint64Stat) String() string {
+	return strconv.FormatUint(s.Value(), 10)
+}
+
+// Set updates the existing value. This satisfies the AtomicUint interface but code can prefer SgwUint64Stat.Store to match atomic.Uint64.
+func (s *SgwUint64Stat) Set(value uint64) {
+	s.Store(value)
+}
+
+func (s *SgwUint64Stat) SetIfMax(value uint64) {
+	for {
+		cur := s.Load()
+		if cur >= value {
+			return
+		}
+
+		if s.CompareAndSwap(cur, value) {
+			return
+		}
+	}
+}
+
+// Value returns the value. This satisfies the AtomicUint interface but new code should prefer SgwUint64Stat.Load to match atomic.Uint64 interface.
+func (s *SgwUint64Stat) Value() uint64 {
+	return s.Load()
+}
+
 func NewFloatStat(subsystem, key, unit, description, addedVersion, deprecatedVersion, stability string, labelKeys, labelVals []string, statValueType prometheus.ValueType, initialValue float64) (*SgwFloatStat, error) {
 	stat, err := newSGWStat(subsystem, key, unit, description, addedVersion, deprecatedVersion, stability, labelKeys, labelVals, statValueType)
 	if err != nil {
@@ -1387,11 +1461,11 @@ func (d *DbStats) initCacheStats() error {
 	if err != nil {
 		return err
 	}
-	resUtil.HighSeqCached, err = NewIntStat(SubsystemCacheKey, "high_seq_cached", StatUnitNoUnits, HighSeqCachedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.HighSeqCached, err = NewUIntStat(SubsystemCacheKey, "high_seq_cached", StatUnitNoUnits, HighSeqCachedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
-	resUtil.HighSeqStable, err = NewIntStat(SubsystemCacheKey, "high_seq_stable", StatUnitNoUnits, HighStableSeqCachedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.HighSeqStable, err = NewUIntStat(SubsystemCacheKey, "high_seq_stable", StatUnitNoUnits, HighStableSeqCachedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
@@ -1684,11 +1758,11 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
-	resUtil.CompactionAttachmentStartTime, err = NewIntStat(SubsystemDatabaseKey, "compaction_attachment_start_time", StatUnitUnixTimestamp, CompactionAttachmentStartTimeDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
+	resUtil.CompactionAttachmentStartTime, err = NewUIntStat(SubsystemDatabaseKey, "compaction_attachment_start_time", StatUnitUnixTimestamp, CompactionAttachmentStartTimeDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
 	if err != nil {
 		return err
 	}
-	resUtil.CompactionTombstoneStartTime, err = NewIntStat(SubsystemDatabaseKey, "compaction_tombstone_start_time", StatUnitUnixTimestamp, CompactionTombstoneStartTimeDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
+	resUtil.CompactionTombstoneStartTime, err = NewUIntStat(SubsystemDatabaseKey, "compaction_tombstone_start_time", StatUnitUnixTimestamp, CompactionTombstoneStartTimeDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.GaugeValue, 0)
 	if err != nil {
 		return err
 	}
@@ -1728,7 +1802,7 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
-	resUtil.HighSeqFeed, err = NewIntStat(SubsystemDatabaseKey, "high_seq_feed", StatUnitNoUnits, HighSeqFeedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.HighSeqFeed, err = NewUIntStat(SubsystemDatabaseKey, "high_seq_feed", StatUnitNoUnits, HighSeqFeedDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
@@ -1776,11 +1850,11 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
-	resUtil.SequenceAssignedCount, err = NewIntStat(SubsystemDatabaseKey, "sequence_assigned_count", StatUnitNoUnits, SequenceAssignedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.SequenceAssignedCount, err = NewUIntStat(SubsystemDatabaseKey, "sequence_assigned_count", StatUnitNoUnits, SequenceAssignedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
-	resUtil.LastSequenceAssignedValue, err = NewIntStat(SubsystemDatabaseKey, "last_sequence_assigned_value", StatUnitNoUnits, LastSequenceAssignedValueDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.LastSequenceAssignedValue, err = NewUIntStat(SubsystemDatabaseKey, "last_sequence_assigned_value", StatUnitNoUnits, LastSequenceAssignedValueDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
@@ -1792,15 +1866,15 @@ func (d *DbStats) initDatabaseStats() error {
 	if err != nil {
 		return err
 	}
-	resUtil.SequenceReleasedCount, err = NewIntStat(SubsystemDatabaseKey, "sequence_released_count", StatUnitNoUnits, SequenceReleasedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.SequenceReleasedCount, err = NewUIntStat(SubsystemDatabaseKey, "sequence_released_count", StatUnitNoUnits, SequenceReleasedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
-	resUtil.SequenceReservedCount, err = NewIntStat(SubsystemDatabaseKey, "sequence_reserved_count", StatUnitNoUnits, SequenceReservedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.SequenceReservedCount, err = NewUIntStat(SubsystemDatabaseKey, "sequence_reserved_count", StatUnitNoUnits, SequenceReservedCountDesc, StatAddedVersion3dot0dot0, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}
-	resUtil.LastSequenceReservedValue, err = NewIntStat(SubsystemDatabaseKey, "last_sequence_reserved_value", StatUnitNoUnits, LastSequenceReservedValueDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
+	resUtil.LastSequenceReservedValue, err = NewUIntStat(SubsystemDatabaseKey, "last_sequence_reserved_value", StatUnitNoUnits, LastSequenceReservedValueDesc, StatAddedVersion3dot2dot4, StatDeprecatedVersionNotDeprecated, StatStabilityCommitted, labelKeys, labelVals, prometheus.CounterValue, 0)
 	if err != nil {
 		return err
 	}

base/util.go

@@ -476,7 +476,7 @@ func RetryLoop[T any](ctx context.Context, description string, worker RetryWorke
 			} else if errors.Is(ctxErr, context.DeadlineExceeded) {
 				verb = "timed out"
 			}
-			return fmt.Errorf("Retry loop for %v %s based on context: %w", description, verb, ctxErr), *new(T)
+			return fmt.Errorf("Retry loop for %v %s based on context: %w", description, verb, context.Cause(ctx)), *new(T)
 		case <-time.After(time.Millisecond * time.Duration(sleepMs)):
 		}
 

base/util_test.go

@@ -239,6 +239,29 @@ func TestRetryLoopContextCancellation(t *testing.T) {
 	}()
 	err, _ := RetryLoop(ctx, t.Name(), worker, CreateIndefiniteMaxDoublingSleeperFunc(10, 100))
 	require.ErrorContains(t, err, "canceled based on context")
+	require.ErrorIs(t, err, context.Canceled)
+	require.Equal(t, ctx.Err(), context.Canceled)
+}
+
+func TestRetryLoopContextCancellationWithCause(t *testing.T) {
+	var invoked atomic.Bool
+	worker := func() (shouldRetry bool, err error, value any) {
+		invoked.Store(true)
+		return true, nil, nil
+	}
+
+	ctx, cancelFunc := context.WithCancelCause(TestCtx(t))
+
+	cause := errors.New("this specific issue")
+	go func() {
+		require.Eventually(t, func() bool { return invoked.Load() == true }, 10*time.Second, 50*time.Millisecond)
+		cancelFunc(cause)
+	}()
+	err, _ := RetryLoop(ctx, t.Name(), worker, CreateIndefiniteMaxDoublingSleeperFunc(10, 100))
+	require.ErrorContains(t, err, "Retry loop for TestRetryLoopContextCancellationWithCause canceled based on context: this specific issue")
+	require.Equal(t, ctx.Err(), context.Canceled)
+	require.ErrorIs(t, err, cause)
+	require.NotErrorIs(t, err, context.Canceled)
 }
 
 func TestRetryLoopContextDeadline(t *testing.T) {
@@ -253,6 +276,8 @@ func TestRetryLoopContextDeadline(t *testing.T) {
 
 	err, _ := RetryLoop(ctx, t.Name(), worker, CreateIndefiniteMaxDoublingSleeperFunc(10, 100))
 	require.ErrorContains(t, err, "timed out based on context")
+	require.ErrorIs(t, err, context.DeadlineExceeded)
+	require.ErrorIs(t, ctx.Err(), context.DeadlineExceeded)
 }
 
 func TestSyncSourceFromURL(t *testing.T) {

base/util_testing.go

@@ -1011,3 +1011,8 @@ func RequireXattrNotFound(t testing.TB, dataStore sgbucket.DataStore, docID stri
 	require.Error(t, err, fmt.Sprintf("Expected xattr %q to not be found on document %q but has contents %s", xattrName, docID, xattrs[xattrName]))
 	RequireXattrNotFoundError(t, err)
 }
+
+// underGoTest returns true if the tests are being run via 'go test'
+func underGoTest() bool {
+	return testing.Testing()
+}