MB-70453: Add random connection support and improve error handling

- Add random bool parameter to getVbConnection() to support random
  connection selection for bucket-level operations that don't require
  vbucket-specific routing
- Enhance processOpError() to handle isAddrNotAvailable and
  isSeveredConnectionError with proper retry logic and connection
  discarding
- Add isSeveredConnectionError() helper function to detect closed/severed
  connection errors
- Refactor GetCollectionsManifest() to use getVbConnection() with random=true
  and leverage existing retry infrastructure via processOpError()
- Update all getVbConnection() call sites to pass random parameter:
  - GetCollectionCID: random=true (bucket-level operation)
  - GetCollectionsManifest: random=true (bucket-level operation)
  - All other operations: random=false (vbucket-specific)
- Fix getRandomConnection() to return errNoPool instead of formatted error
  for consistency with error handling

Change-Id: I06d474603aea1360d4fb2e286e6ecd65693f0b3d
Reviewed-on: https://review.couchbase.org/c/query/+/240776
Well-Formed: Restriction Checker
Reviewed-by: Dhanya Gowrish <dhanya.gowrish@couchbase.com>
Tested-by: Sitaram Vemulapalli <sitaram.vemulapalli@couchbase.com>

Author: Sohoxic

primitives/couchbase/memcached.go

@@ -150,12 +150,17 @@ func (b *Bucket) getConnectionToVBucket(vb uint32, desc *doDescriptor) (*memcach
 }
 
 // first part of the retry loop: get a connection and handle errors
-func (b *Bucket) getVbConnection(vb uint32, desc *doDescriptor) (conn *memcached.Client, pool *connectionPool, err error) {
+// The collections manifest is a bucket-level resource available from any KV node, so we use getVbConnection
+// with random=true to pick an arbitrary node while still benefiting from its built-in retry and error-handling logic.
+func (b *Bucket) getVbConnection(vb uint32, desc *doDescriptor, random bool) (conn *memcached.Client, pool *connectionPool, err error) {
 	desc.errorString = ""
 
-	// if we had a NMVB and have successfully identified the pool for the correct node, use it
-	// if it doesn't work out, fall back to the old method
-	if desc.pool != nil {
+	if random {
+		desc.pool = nil
+		conn, pool, err = b.getRandomConnection()
+	} else if desc.pool != nil {
+		// if we had a NMVB and have successfully identified the pool for the correct node, use it
+		// if it doesn't work out, fall back to the old method
 		pool = desc.pool
 		desc.pool = nil
 		conn, err = pool.Get()
@@ -360,6 +365,17 @@ func (b *Bucket) processOpError(vb uint32, lastError error, node string, desc *d
 			desc.discard = true
 			desc.backOffAttempts++
 			desc.retry = backOff(desc.backOffAttempts, desc.maxTries, backOffDuration, true)
+		} else if isAddrNotAvailable(lastError) {
+			desc.discard = true
+			desc.backOffAttempts++
+			desc.retry = backOff(desc.backOffAttempts, desc.maxTries, backOffDuration, true)
+			if desc.retry {
+				b.Refresh()
+			}
+		} else if isSeveredConnectionError(lastError) {
+			desc.discard = true
+			desc.backOffAttempts++
+			desc.retry = backOff(desc.backOffAttempts, desc.maxTries, backOffDuration, true)
 		} else if IsReadTimeOutError(lastError) {
 			desc.discard = true
 			desc.retry = true
@@ -460,25 +476,26 @@ func (b *Bucket) backOffRetries() int {
 // Note that this automatically handles transient errors by replaying
 // your function on a (for example) "not-my-vbucket" error, so don't assume your command will only be executed once.
 func (b *Bucket) do(k string, f func(mc *memcached.Client, vb uint16) error) (err error) {
-	return b.do2(k, f, true, false, b.backOffRetries())
+	return b.do2(k, f, true, false, false, b.backOffRetries())
 }
 
-func (b *Bucket) do2(k string, f func(mc *memcached.Client, vb uint16) error, deadline bool, useReplicas bool,
+func (b *Bucket) do2(k string, f func(mc *memcached.Client, vb uint16) error, deadline, useReplicas, random bool,
 	backOffRetries int) (err error) {
 
 	vb := b.VBHash(k)
 
-	return b.do3(uint16(vb), f, deadline, useReplicas, backOffRetries)
+	err = b.do3(uint16(vb), f, deadline, useReplicas, random, backOffRetries)
+	return err
 }
 
-func (b *Bucket) do3(vb uint16, f func(mc *memcached.Client, vb uint16) error, deadline bool, useReplicas bool,
+func (b *Bucket) do3(vb uint16, f func(mc *memcached.Client, vb uint16) error, deadline, useReplicas, random bool,
 	backOffRetries int) (err error) {
 
 	var lastError error
 
 	desc := &doDescriptor{useReplicas: useReplicas, version: b.Version, maxTries: backOffRetries}
 	for desc.attempts = 0; desc.attempts < desc.maxTries; desc.attempts++ {
-		conn, pool, err := b.getVbConnection(uint32(vb), desc)
+		conn, pool, err := b.getVbConnection(uint32(vb), desc, random)
 		if err != nil {
 			if desc.retry {
 				continue
@@ -751,6 +768,20 @@ func isAddrNotAvailable(err error) bool {
 	return strings.Contains(estr, "cannot assign requested address")
 }
 
+func isSeveredConnectionError(err error) bool {
+	if err == nil {
+		return false
+	}
+	if err == io.EOF {
+		return true
+	}
+	estr := err.Error()
+	return strings.Contains(estr, "use of closed network connection") ||
+		strings.Contains(estr, "connection closed") ||
+		strings.Contains(estr, "broken pipe") ||
+		strings.Contains(estr, "connection reset")
+}
+
 // Get the deadline for your connection
 func getDeadline(reqDeadline time.Time, kvTimeout time.Duration, duration time.Duration) (time.Time, error) {
 
@@ -833,7 +864,7 @@ func (b *Bucket) doBulkGet(vb uint16, keys []string, active func() bool, reqDead
 		// This stack frame exists to ensure we can clean up
 		// connection at a reasonable time.
 		err := func() error {
-			conn, pool, err := b.getVbConnection(uint32(vb), desc)
+			conn, pool, err := b.getVbConnection(uint32(vb), desc, false)
 			if err != nil {
 				if !desc.retry {
 					if desc.errorString != "" {
@@ -1405,7 +1436,7 @@ func (b *Bucket) GetCollectionCID(scope string, collection string, reqDeadline t
 		collUid = binary.BigEndian.Uint32(response.Extras[8:12])
 
 		return nil
-	}, false, false, b.backOffRetries())
+	}, false, false, true, b.backOffRetries())
 
 	return collUid, manifestUid, err
 }
@@ -1440,7 +1471,7 @@ func (b *Bucket) GetsMC(key string, active func() bool, reqDeadline time.Time, k
 			return err1
 		}
 		return nil
-	}, false, useReplica, b.backOffRetries())
+	}, false, useReplica, false, b.backOffRetries())
 	atomic.AddUint64(&b.readCount, 1)
 	return response, err
 }
@@ -1467,7 +1498,7 @@ func (b *Bucket) GetsSubDoc(key string, reqDeadline time.Time, kvTimeout time.Du
 		mc.SetDeadline(dl)
 		response, err1 = mc.GetSubdoc(vb, key, paths, context...)
 		return err1
-	}, false, false, b.backOffRetries())
+	}, false, false, false, b.backOffRetries())
 	return response, err
 }
 
@@ -1487,7 +1518,7 @@ func (b *Bucket) SetsSubDoc(key string, ops []memcached.SubDocOp, context ...*me
 		mc.SetDeadline(noDeadline)
 		response, err = mc.SetSubdoc(vb, key, ops, context...)
 		return err
-	}, false, false, b.backOffRetries())
+	}, false, false, false, b.backOffRetries())
 	if err == nil && err1 != nil {
 		err = err1
 	}
@@ -1500,7 +1531,7 @@ func (b *Bucket) getRandomConnection() (*memcached.Client, *connectionPool, erro
 		var currentPool = 0
 		pools := b.getConnPools(false /* not already locked */)
 		if len(pools) == 0 {
-			return nil, nil, fmt.Errorf("No connection pool found")
+			return nil, nil, errNoPool
 		} else if len(pools) > 1 { // choose a random connection
 			currentPool = rand.Intn(len(pools))
 		} // if only one pool, currentPool defaults to 0, i.e., the only pool

primitives/couchbase/memcached_scan.go

@@ -1738,7 +1738,7 @@ func (cqueue *rswCancelQueue) runWorker() {
 						vbucket = cr[i].vbucket
 						desc := &doDescriptor{useReplicas: true, version: b.Version, maxTries: b.backOffRetries(), retry: true}
 						for desc.attempts = 0; desc.attempts < desc.maxTries; {
-							conn, pool, err = b.getVbConnection(uint32(vbucket), desc)
+							conn, pool, err = b.getVbConnection(uint32(vbucket), desc, false)
 							if err != nil {
 								if desc.retry {
 									desc.attempts++
@@ -1914,7 +1914,7 @@ func (queue *rswQueue) runWorker() {
 					desc := &doDescriptor{useReplicas: vbscan.scan.useReplica, version: b.Version, maxTries: b.backOffRetries(),
 						retry: true}
 					for desc.attempts = 0; desc.attempts < desc.maxTries; {
-						conn, pool, err = b.getVbConnection(uint32(vb), desc)
+						conn, pool, err = b.getVbConnection(uint32(vb), desc, false)
 						if err != nil {
 							if desc.retry {
 								desc.attempts++

primitives/couchbase/ns_server.go

@@ -1108,47 +1108,81 @@ func (b *Bucket) GetCollectionsManifest() (*Manifest, error) {
 		return nil, fmt.Errorf("Collections not enabled.")
 	}
 
-	b.RLock()
-	pools := b.getConnPools(true /* already locked */)
-	if len(pools) == 0 {
-		b.RUnlock()
-		return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest: no connection pool. No collections "+
-			"access to bucket %s.", b.Name)
-	}
-	pool := pools[0] // Any pool will do, so use the first one.
-	b.RUnlock()
-	client, err := pool.Get()
-	if err != nil {
-		return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest: %v. No collections access to "+
-			"bucket %s.", err, b.Name)
-	}
-	dl, _ := getDeadline(noDeadline, _NO_TIMEOUT, DefaultTimeout)
-	client.SetDeadline(dl)
-
-	// We need to select the bucket before GetCollectionsManifest()
-	// will work. This is sometimes done at startup (see defaultMkConn())
-	// but not always, depending on the auth type.
-	// Doing this is safe because we collect the the connections
-	// by bucket, so the bucket being selected will never change.
-	_, err = client.SelectBucket(b.Name)
-	if err != nil {
-		pool.Return(client)
-		return nil, fmt.Errorf("Unable to select bucket %s: %v. No collections access to bucket %s.", err, b.Name, b.Name)
-	}
+	vb := uint32(b.VBHash("DUMMY")) // any key
+	desc := &doDescriptor{version: b.Version, maxTries: b.backOffRetries()}
+	var res *gomemcached.MCResponse
+	var lastError error
+	selectFailed := false
+	for desc.attempts = 0; desc.attempts < desc.maxTries; desc.attempts++ {
+		// The collections manifest is a bucket-level resource available from any KV node,
+		// so we use getVbConnection with random=true to pick an arbitrary node while still
+		// benefiting from its built-in retry and error-handling logic.
+		conn, pool, err := b.getVbConnection(vb, desc, true)
+		if err != nil {
+			lastError = err
+			if desc.retry {
+				continue
+			}
+			if desc.errorString != "" {
+				return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest for bucket %s: %v",
+					b.Name, fmt.Errorf(desc.errorString, b.Name, err))
+			}
+			return nil, fmt.Errorf("Unable to get connection to retrieve collections manifest for bucket %s: %v",
+				b.Name, err)
+		}
 
-	res, err := client.GetCollectionsManifest()
-	if err != nil {
-		pool.Return(client)
-		return nil, fmt.Errorf("Unable to retrieve collections manifest: %v. No collections access to bucket %s.", err, b.Name)
+		dl, _ := getDeadline(noDeadline, _NO_TIMEOUT, DefaultTimeout)
+		conn.SetDeadline(dl)
+		// We need to select the bucket before GetCollectionsManifest()
+		// will work. This is sometimes done at startup (see defaultMkConn())
+		// but not always, depending on the auth type.
+		// Doing this is safe because we collect the the connections
+		// by bucket, so the bucket being selected will never change.
+		selectFailed = false
+		_, lastError = conn.SelectBucket(b.Name)
+		if lastError == nil {
+			res, lastError = conn.GetCollectionsManifest()
+		} else {
+			selectFailed = true
+		}
+		b.processOpError(vb, lastError, pool.Node(), desc)
+		if desc.discard {
+			pool.Discard(conn)
+		} else {
+			conn.SetReplica(false)
+			pool.Return(conn)
+		}
+		if lastError == nil {
+			mani, err := parseCollectionsManifest(res)
+			if err != nil {
+				return nil, fmt.Errorf("Unable to parse collections manifest: %v. No collections access to bucket %s.",
+					err, b.Name)
+			}
+			return mani, nil
+		} else if !desc.retry {
+			if selectFailed {
+				return nil, fmt.Errorf("Unable to select bucket %s: %v. No collections access to bucket %s.",
+					b.Name, lastError, b.Name)
+			}
+
+			break
+		}
 	}
-	mani, err := parseCollectionsManifest(res)
-	if err != nil {
-		pool.Return(client)
-		return nil, fmt.Errorf("Unable to parse collections manifest: %v. No collections access to bucket %s.", err, b.Name)
+	if lastError != nil {
+		if selectFailed {
+			return nil, fmt.Errorf("Unable to select bucket %s: %v. No collections access to bucket %s.",
+				b.Name, lastError, b.Name)
+		}
+		if desc.errorString != "" {
+			return nil, fmt.Errorf("Unable to retrieve collections manifest: %v. No collections access to bucket %s.",
+				fmt.Errorf(desc.errorString, b.Name, lastError), b.Name)
+		}
+		return nil, fmt.Errorf("Unable to retrieve collections manifest: %v after %d attempts. No collections access to bucket %s.", lastError, desc.attempts, b.Name)
 	}
-
-	pool.Return(client)
-	return mani, nil
+	// The final return here is only reachable when lastError == nil after the loop exits. That can only happen if desc.maxTries is 0 — the loop condition
+	// desc.attempts < desc.maxTries is immediately false, the body never executes, and lastError stays at its zero value (nil). This is just a defensive
+	// guard for the maxTries == 0 edge case and also satisfies Go's requirement that all code paths return a value.
+	return nil, fmt.Errorf("Unable to retrieve collections manifest after %d attempts. No collections access to bucket %s.", desc.attempts, b.Name)
 }
 
 func (b *Bucket) RefreshFully() error {