[3.2.2 backport] CBG-4446 create a cancel context inside BlipSyncContext (#7314)

* CBG-4370 preqreq deduplicate active replicator pull changes (#7211)

* CBG-4370 create a cancel context inside BlipSyncContext (#7201)

* CBG-4370 create a cancel context inside BlipSyncContext

This cancel context allows a forceable closure of the underlying blip
connection.

In the case there is a continuous pull replication and there is an error
on the changes feed, the only way to stop the pull replication is to
shut down the connection. CBL clients do not listen to unsolicited error
messages.

* Avoid refactoring with the change

* pass lint

* test active replicator reconnection

* test fixup

* lint correctly

* PR comments

* Acquire read lock to present data race

* switch query to use all docs index

* fix lint issue

Author: torcolvin

base/leaky_bucket.go

@@ -125,7 +125,12 @@ type LeakyBucketConfig struct {
 
 	// Returns a partial error the first time ViewCustom is called
 	FirstTimeViewCustomPartialError bool
-	PostQueryCallback               func(ddoc, viewName string, params map[string]interface{}) // Issues callback after issuing query when bucket.ViewQuery is called
+
+	// QueryCallback allows tests to set a callback that will be issued prior to issuing a view query
+	QueryCallback     func(ddoc, viewName string, params map[string]any) error
+	PostQueryCallback func(ddoc, viewName string, params map[string]interface{}) // Issues callback after issuing query when bucket.ViewQuery is called
+
+	N1QLQueryCallback func(ctx context.Context, statement string, params map[string]any, consistency ConsistencyMode, adhoc bool) error
 
 	PostN1QLQueryCallback func()
 

base/leaky_datastore.go

@@ -241,6 +241,12 @@ func (lds *LeakyDataStore) ViewQuery(ctx context.Context, ddoc, name string, par
 	if !ok {
 		return nil, errors.New("bucket does not support views")
 	}
+	if lds.config.QueryCallback != nil {
+		err := lds.config.QueryCallback(ddoc, name, params)
+		if err != nil {
+			return nil, err
+		}
+	}
 	iterator, err := vs.ViewQuery(ctx, ddoc, name, params)
 
 	if lds.config.FirstTimeViewCustomPartialError {
@@ -324,10 +330,14 @@ func (lds *LeakyDataStore) SetFirstTimeViewCustomPartialError(val bool) {
 	lds.config.FirstTimeViewCustomPartialError = val
 }
 
-func (lds *LeakyDataStore) SetPostQueryCallback(callback func(ddoc, viewName string, params map[string]interface{})) {
+func (lds *LeakyDataStore) SetPostQueryCallback(callback func(ddoc, viewName string, params map[string]any)) {
 	lds.config.PostQueryCallback = callback
 }
 
+func (lds *LeakyDataStore) SetQueryCallback(fn func(ddoc, viewName string, params map[string]any) error) {
+	lds.config.QueryCallback = fn
+}
+
 func (lds *LeakyDataStore) SetPostN1QLQueryCallback(callback func()) {
 	lds.config.PostN1QLQueryCallback = callback
 }
@@ -447,6 +457,12 @@ func (lds *LeakyDataStore) Query(ctx context.Context, statement string, params m
 	if err != nil {
 		return nil, err
 	}
+	if lds.config.N1QLQueryCallback != nil {
+		err := lds.config.N1QLQueryCallback(ctx, statement, params, consistency, adhoc)
+		if err != nil {
+			return nil, err
+		}
+	}
 	iterator, err := n1qlStore.Query(ctx, statement, params, consistency, adhoc)
 
 	if lds.config.PostN1QLQueryCallback != nil {

db/active_replicator.go

@@ -208,9 +208,10 @@ func connect(arc *activeReplicatorCommon, idSuffix string) (blipSender *blip.Sen
 	arc.replicationStats.NumConnectAttempts.Add(1)
 
 	var originPatterns []string // no origin headers for ISGR
-	// NewSGBlipContext doesn't set cancellation context - active replication cancellation on db close is handled independently
-	blipContext, err := NewSGBlipContext(arc.ctx, arc.config.ID+idSuffix, originPatterns, nil)
+	cancelCtx, cancelFunc := context.WithCancel(context.Background())
+	blipContext, err := NewSGBlipContext(arc.ctx, arc.config.ID+idSuffix, originPatterns, cancelCtx)
 	if err != nil {
+		cancelFunc()
 		return nil, nil, err
 	}
 	blipContext.WebsocketPingInterval = arc.config.WebsocketPingInterval
@@ -221,7 +222,10 @@ func connect(arc *activeReplicatorCommon, idSuffix string) (blipSender *blip.Sen
 		}
 	}
 
-	bsc = NewBlipSyncContext(arc.ctx, blipContext, arc.config.ActiveDB, blipContext.ID, arc.replicationStats)
+	bsc, err = NewBlipSyncContext(arc.ctx, blipContext, arc.config.ActiveDB, blipContext.ID, arc.replicationStats, cancelFunc)
+	if err != nil {
+		return nil, nil, err
+	}
 
 	bsc.loggingCtx = base.CorrelationIDLogCtx(
 		arc.config.ActiveDB.AddDatabaseLogContext(base.NewNonCancelCtx().Ctx),

db/active_replicator_common.go

@@ -15,6 +15,7 @@ import (
 	"errors"
 	"expvar"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -52,7 +53,7 @@ type activeReplicatorCommon struct {
 	ctxCancel             context.CancelFunc
 	reconnectActive       base.AtomicBool                                             // Tracks whether reconnect goroutine is active
 	replicatorConnectFn   func() error                                                // the function called inside reconnectLoop.
-	activeSendChanges     base.AtomicBool                                             // Tracks whether sendChanges goroutine is active.
+	activeSendChanges     atomic.Int32                                                // Tracks whether sendChanges goroutines are active, there is one per collection.
 	namedCollections      map[base.ScopeAndCollectionName]*activeReplicatorCollection // set only if the replicator is running with collections - access with forEachCollection
 	defaultCollection     *activeReplicatorCollection                                 // set only if the replicator is not running with collections - access with forEachCollection
 }
@@ -312,8 +313,9 @@ func (a *activeReplicatorCommon) getState() string {
 	return a.state
 }
 
-// requires a.stateErrorLock
 func (a *activeReplicatorCommon) _getStateWithErrorMessage() (state string, lastErrorMessage string) {
+	a.stateErrorLock.RLock()
+	defer a.stateErrorLock.RUnlock()
 	if a.lastError == nil {
 		return a.state, ""
 	}
@@ -356,6 +358,14 @@ func (a *activeReplicatorCommon) getCheckpointHighSeq() string {
 	return highSeqStr
 }
 
+// publishStatus updates the replication status document in the metadata store.
+func (a *activeReplicatorCommon) publishStatus() {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a._publishStatus()
+}
+
+// _publishStatus updates the replication status document in the metadata store. Requires holding a.lock before calling.
 func (a *activeReplicatorCommon) _publishStatus() {
 	status := a._getStatusCallback()
 	err := setLocalStatus(a.ctx, a.config.ActiveDB.MetadataStore, a.statusKey, status, int(a.config.ActiveDB.Options.LocalDocExpirySecs))

db/active_replicator_push.go

@@ -268,7 +268,7 @@ func (apr *ActivePushReplicator) Stop() error {
 		return err
 	}
 	teardownStart := time.Now()
-	for apr.activeSendChanges.IsTrue() && (time.Since(teardownStart) < time.Second*10) {
+	for apr.activeSendChanges.Load() != 0 && (time.Since(teardownStart) < time.Second*10) {
 		time.Sleep(10 * time.Millisecond)
 	}
 	return nil
@@ -298,8 +298,17 @@ func (apr *ActivePushReplicator) _startPushNonCollection() error {
 	bh.collection = dbCollectionWithUser
 	bh.loggingCtx = bh.collection.AddCollectionContext(bh.BlipSyncContext.loggingCtx)
 
+	return apr._startSendingChanges(bh, apr.defaultCollection.Checkpointer.lastCheckpointSeq)
+}
+
+// _startSendingChanges starts a changes feed for a given collection in a goroutine and starts sending changes to the passive peer from a starting sequence value.
+func (apr *ActivePushReplicator) _startSendingChanges(bh *blipHandler, since SequenceID) error {
+	collectionCtx, err := bh.collections.get(bh.collectionIdx)
+	if err != nil {
+		return err
+	}
 	var channels base.Set
-	if filteredChannels := apr.config.getFilteredChannels(nil); len(filteredChannels) > 0 {
+	if filteredChannels := apr.config.getFilteredChannels(bh.collectionIdx); len(filteredChannels) > 0 {
 		channels = base.SetFromArray(filteredChannels)
 	}
 
@@ -320,17 +329,12 @@ func (apr *ActivePushReplicator) _startPushNonCollection() error {
 		// No special handling for error
 	}
 
-	collectionCtx, err := bh.collections.get(nil)
-	if err != nil {
-		return err
-	}
-
-	apr.activeSendChanges.Set(true)
+	apr.activeSendChanges.Add(1)
 	go func(s *blip.Sender) {
-		defer apr.activeSendChanges.Set(false)
-		isComplete := bh.sendChanges(s, &sendChangesOptions{
+		defer apr.activeSendChanges.Add(-1)
+		isComplete, err := bh.sendChanges(s, &sendChangesOptions{
 			docIDs:            apr.config.DocIDs,
-			since:             apr.defaultCollection.Checkpointer.lastCheckpointSeq,
+			since:             since,
 			continuous:        apr.config.Continuous,
 			activeOnly:        apr.config.ActiveOnly,
 			batchSize:         int(apr.config.ChangesBatchSize),
@@ -340,11 +344,17 @@ func (apr *ActivePushReplicator) _startPushNonCollection() error {
 			ignoreNoConflicts: true, // force the passive side to accept a "changes" message, even in no conflicts mode.
 			changesCtx:        collectionCtx.changesCtx,
 		})
-		// On a normal completion, call complete for the replication
+		if err != nil {
+			base.InfofCtx(apr.ctx, base.KeyReplicate, "Terminating blip connection due to changes feed error: %v", err)
+			bh.ctxCancelFunc()
+			_ = apr.setError(err)
+			apr.publishStatus()
+			return
+		}
 		if isComplete {
+			// On a normal completion, call complete for the replication
 			apr.Complete()
 		}
 	}(apr.blipSender)
-
 	return nil
 }

db/active_replicator_push_collections.go

@@ -10,9 +10,7 @@ package db
 
 import (
 	"fmt"
-	"strings"
 
-	"github.com/couchbase/go-blip"
 	"github.com/couchbase/sync_gateway/base"
 )
 
@@ -52,47 +50,6 @@ func (apr *ActivePushReplicator) _startPushWithCollections() error {
 		bh.collectionIdx = collectionIdx
 		bh.loggingCtx = bh.collection.AddCollectionContext(bh.BlipSyncContext.loggingCtx)
 
-		var channels base.Set
-		if filteredChannels := apr.config.getFilteredChannels(collectionIdx); len(filteredChannels) > 0 {
-			channels = base.SetFromArray(filteredChannels)
-		}
-
-		apr.blipSyncContext.fatalErrorCallback = func(err error) {
-			if strings.Contains(err.Error(), ErrUseProposeChanges.Message) {
-				err = ErrUseProposeChanges
-				_ = apr.setError(PreHydrogenTargetAllowConflictsError)
-				err = apr.stopAndDisconnect()
-				if err != nil {
-					base.ErrorfCtx(apr.ctx, "Failed to stop and disconnect replication: %v", err)
-				}
-			} else if strings.Contains(err.Error(), ErrDatabaseWentAway.Message) {
-				err = apr.reconnect()
-				if err != nil {
-					base.ErrorfCtx(apr.ctx, "Failed to reconnect replication: %v", err)
-				}
-			}
-			// No special handling for error
-		}
-		apr.activeSendChanges.Set(true)
-		go func(s *blip.Sender) {
-			defer apr.activeSendChanges.Set(false)
-			isComplete := bh.sendChanges(s, &sendChangesOptions{
-				docIDs:            apr.config.DocIDs,
-				since:             replicationCollection.Checkpointer.lastCheckpointSeq,
-				continuous:        apr.config.Continuous,
-				activeOnly:        apr.config.ActiveOnly,
-				batchSize:         int(apr.config.ChangesBatchSize),
-				revocations:       apr.config.PurgeOnRemoval,
-				channels:          channels,
-				clientType:        clientTypeSGR2,
-				ignoreNoConflicts: true, // force the passive side to accept a "changes" message, even in no conflicts mode.
-				changesCtx:        c.changesCtx,
-			})
-			// On a normal completion, call complete for the replication
-			if isComplete {
-				apr.Complete()
-			}
-		}(apr.blipSender)
-		return nil
+		return apr._startSendingChanges(bh, replicationCollection.Checkpointer.lastCheckpointSeq)
 	})
 }

db/blip_handler.go

@@ -348,7 +348,7 @@ func (bh *blipHandler) handleSubChanges(rq *blip.Message) error {
 		}()
 		// sendChanges runs until blip context closes, or fails due to error
 		startTime := time.Now()
-		_ = bh.sendChanges(rq.Sender, &sendChangesOptions{
+		_, err = bh.sendChanges(rq.Sender, &sendChangesOptions{
 			docIDs:            subChangesParams.docIDs(),
 			since:             subChangesParams.Since(),
 			continuous:        continuous,
@@ -361,6 +361,10 @@ func (bh *blipHandler) handleSubChanges(rq *blip.Message) error {
 			changesCtx:        collectionCtx.changesCtx,
 			requestPlusSeq:    requestPlusSeq,
 		})
+		if err != nil {
+			base.DebugfCtx(bh.loggingCtx, base.KeySyncMsg, "Closing blip connection due to changes feed error %+v\n", err)
+			bh.ctxCancelFunc()
+		}
 		base.DebugfCtx(bh.loggingCtx, base.KeySyncMsg, "#%d: Type:%s   --> Time:%v", bh.serialNumber, rq.Profile(), time.Since(startTime))
 	}()
 
@@ -428,8 +432,8 @@ func (flag changesDeletedFlag) HasFlag(deletedFlag changesDeletedFlag) bool {
 	return flag&deletedFlag != 0
 }
 
-// Sends all changes since the given sequence
-func (bh *blipHandler) sendChanges(sender *blip.Sender, opts *sendChangesOptions) (isComplete bool) {
+// sendChanges will start a changes feed and send changes. Returns bool to indicate whether the changes feed finished and all changes were sent. The error value is only used to indicate a fatal error, where the blip connection should be terminated. If the blip connection is disconnected by the client, the error will be nil, but the boolean parameter will be false.
+func (bh *blipHandler) sendChanges(sender *blip.Sender, opts *sendChangesOptions) (bool, error) {
 	defer func() {
 		if panicked := recover(); panicked != nil {
 			bh.replicationStats.NumHandlersPanicked.Add(1)
@@ -472,11 +476,10 @@ func (bh *blipHandler) sendChanges(sender *blip.Sender, opts *sendChangesOptions
 	changesDb, err := bh.copyDatabaseCollectionWithUser(bh.collectionIdx)
 	if err != nil {
 		base.WarnfCtx(bh.loggingCtx, "[%s] error sending changes: %v", bh.blipContext.ID, err)
-		return false
-
+		return false, err
 	}
 
-	forceClose := generateBlipSyncChanges(bh.loggingCtx, changesDb, channelSet, options, opts.docIDs, func(changes []*ChangeEntry) error {
+	forceClose, err := generateBlipSyncChanges(bh.loggingCtx, changesDb, channelSet, options, opts.docIDs, func(changes []*ChangeEntry) error {
 		base.DebugfCtx(bh.loggingCtx, base.KeySync, "    Sending %d changes", len(changes))
 		for _, change := range changes {
 			if !strings.HasPrefix(change.ID, "_") {
@@ -538,8 +541,7 @@ func (bh *blipHandler) sendChanges(sender *blip.Sender, opts *sendChangesOptions
 		}
 		bh.db.DatabaseContext.NotifyTerminatedChanges(bh.loggingCtx, user)
 	}
-
-	return !forceClose
+	return (err == nil && !forceClose), err
 }
 
 func (bh *blipHandler) buildChangesRow(change *ChangeEntry, revID string) []interface{} {

db/blip_sync_context.go

@@ -34,7 +34,10 @@ const (
 
 var ErrClosedBLIPSender = errors.New("use of closed BLIP sender")
 
-func NewBlipSyncContext(ctx context.Context, bc *blip.Context, db *Database, contextID string, replicationStats *BlipSyncStats) *BlipSyncContext {
+func NewBlipSyncContext(ctx context.Context, bc *blip.Context, db *Database, contextID string, replicationStats *BlipSyncStats, ctxCancelFunc context.CancelFunc) (*BlipSyncContext, error) {
+	if ctxCancelFunc == nil {
+		return nil, errors.New("cancelCtxFunc is required")
+	}
 	maxInFlightChangesBatches := DefaultMaxConcurrentChangesBatches
 	if db.Options.MaxConcurrentChangesBatches != nil {
 		maxInFlightChangesBatches = *db.Options.MaxConcurrentChangesBatches
@@ -55,6 +58,7 @@ func NewBlipSyncContext(ctx context.Context, bc *blip.Context, db *Database, con
 		inFlightChangesThrottle: make(chan struct{}, maxInFlightChangesBatches),
 		inFlightRevsThrottle:    make(chan struct{}, maxInFlightRevs),
 		collections:             &blipCollections{},
+		ctxCancelFunc:           ctxCancelFunc,
 	}
 	if bsc.replicationStats == nil {
 		bsc.replicationStats = NewBlipSyncStats()
@@ -86,7 +90,7 @@ func NewBlipSyncContext(ctx context.Context, bc *blip.Context, db *Database, con
 			bsc.register(profile, handlerFn)
 		}
 	}
-	return bsc
+	return bsc, nil
 }
 
 // BlipSyncContext represents one BLIP connection (socket) opened by a client.
@@ -133,6 +137,8 @@ type BlipSyncContext struct {
 	collections *blipCollections // all collections handled by blipSyncContext, implicit or via GetCollections
 
 	stats blipSyncStats // internal structure to store stats
+
+	ctxCancelFunc context.CancelFunc // function to cancel a blip replication
 }
 
 // blipSyncStats has support structures to support reporting stats at regular interval
@@ -248,6 +254,7 @@ func (bsc *BlipSyncContext) Close() {
 		}
 		bsc.reportStats(true)
 		close(bsc.terminator)
+		bsc.ctxCancelFunc()
 	})
 }