CBG-5049: swap caching mutex from RW to exclusive only

db/change_cache.go

@@ -72,7 +72,7 @@ type changeCache struct {
 	started            base.AtomicBool                     // Set by the Start method
 	stopped            base.AtomicBool                     // Set by the Stop method
 	skippedSeqs        *SkippedSequenceSkiplist            // Skipped sequences still pending on the DCP caching feed
-	lock               sync.RWMutex                        // Coordinates access to struct fields
+	lock               sync.Mutex                          // Coordinates access to struct fields
 	options            CacheOptions                        // Cache config
 	terminator         chan bool                           // Signal termination of background goroutines
 	backgroundTasks    []BackgroundTask                    // List of background tasks.
@@ -93,8 +93,8 @@ type changeCacheStats struct {
 
 func (c *changeCache) updateStats(ctx context.Context) {
 
-	c.lock.RLock()
-	defer c.lock.RUnlock()
+	c.lock.Lock()
+	defer c.lock.Unlock()
 	if c.db == nil {
 		return
 	}
@@ -990,9 +990,9 @@ func (c *changeCache) _setInitialSequence(initialSequence uint64) {
 
 // Concurrent-safe get value of nextSequence
 func (c *changeCache) getNextSequence() (nextSequence uint64) {
-	c.lock.RLock()
+	c.lock.Lock()
 	nextSequence = c.nextSequence
-	c.lock.RUnlock()
+	c.lock.Unlock()
 	return nextSequence
 }
 

db/database.go

@@ -175,7 +175,10 @@ func (db *DatabaseContext) PushSkipped(ctx context.Context, seq uint64) {
 }
 
 func (db *DatabaseContext) RemoveSkipped(seq uint64) {
-	db.changeCache.RemoveSkipped(seq)
+	err := db.changeCache.RemoveSkipped(seq)
+	if err != nil {
+		base.WarnfCtx(context.TODO(), "Error removing skipped sequence %d: %v", seq, err)
+	}
 }
 
 type Scope struct {