Router.Route: handle outdated *Replicaset object (resolve #11)

- Router.Call bugfix: set destinationName := rs.info.Name if destination exists
- Router.Route bugfix: handle outdated *Replicaset object
- make comments more go-style
- Router.cronDiscovery: log panic in another goroutine
- copy maps using copyMap generic

Author: nurzhan-saktaganov

CHANGELOG.md

@@ -1,10 +1,16 @@
 ## Unreleased
 
+BUG FIXES:
+* Router.Call bugfix: set destinationName := rs.info.Name if destination exists.
+* Router.Route bugfix: handle outdated *Replicaset object (resolve issue #11).
+
 FEATURES:
 * Now when calling RemoveInstance, if an empty replicaset name is passed, the replicaset will be calculated automatically.
 
 CHANGES:
 * Bump go-tarantool from v2.2.1 to v2.3.0.
+* Make comments more go-style.
+* Router.cronDiscovery: log panic in another goroutine.
 
 TESTS:
 * Fixed etcd overlapping ports.

api.go

@@ -311,35 +311,43 @@ func (r *Router) Call(ctx context.Context, bucketID uint64, mode CallMode,
 				// We reproduce here behavior in https://github.com/tarantool/vshard/blob/b6fdbe950a2e4557f05b83bd8b846b126ec3724e/vshard/router/init.lua#L663
 				r.BucketReset(bucketID)
 
-				if destination := vshardError.Destination; destination != "" {
+				destination := vshardError.Destination
+				if destination != "" {
 					var loggedOnce bool
 					for {
 						nameToReplicasetRef := r.getNameToReplicaset()
 
-						_, destinationExists := nameToReplicasetRef[destination]
+						// In some cases destination contains UUID (prior to tnt 3.x), in some cases it contains replicaset name.
+						// So, at this point we don't know what destination is: a name or an UUID.
+						// But we need a name to access values in nameToReplicasetRef map, so let's find it out.
+						var destinationName string
 
-						if !destinationExists {
+						_, destinationExists := nameToReplicasetRef[destination]
+						if destinationExists {
+							destinationName = destination
+						} else {
 							// for older logic with uuid we must support backward compatibility
 							// if destination is uuid and not name, lets find it too
-							for _, rsRef := range nameToReplicasetRef {
-								if rsRef.info.UUID.String() == destination {
+							for rsName, rs := range nameToReplicasetRef {
+								if rs.info.UUID.String() == destination {
 									destinationExists = true
+									destinationName = rsName
 									break
 								}
 							}
 						}
 
 						if destinationExists {
-							_, err := r.BucketSet(bucketID, destination)
+							_, err := r.BucketSet(bucketID, destinationName)
 							if err == nil {
 								break // breaks loop
 							}
-							r.log().Warnf(ctx, "Failed set bucket %d to %v (possible race): %v", bucketID, destination, err)
+							r.log().Warnf(ctx, "Failed set bucket %d to %v (possible race): %v", bucketID, destinationName, err)
 						}
 
 						if !loggedOnce {
 							r.log().Warnf(ctx, "Replicaset '%v' was not found, but received from storage as destination - please "+
-								"update configuration", destination)
+								"update configuration", destinationName)
 							loggedOnce = true
 						}
 
@@ -662,11 +670,5 @@ func (r *Router) RouteAll() map[string]*Replicaset {
 	nameToReplicasetRef := r.getNameToReplicaset()
 
 	// Do not expose the original map to prevent unauthorized modification.
-	nameToReplicasetCopy := make(map[string]*Replicaset)
-
-	for k, v := range nameToReplicasetRef {
-		nameToReplicasetCopy[k] = v
-	}
-
-	return nameToReplicasetCopy
+	return copyMap(nameToReplicasetRef)
 }

discovery.go

@@ -55,7 +55,20 @@ func (r *Router) Route(ctx context.Context, bucketID uint64) (*Replicaset, error
 
 	rs := routeMap[bucketID].Load()
 	if rs != nil {
-		return rs, nil
+		nameToReplicasetRef := r.getNameToReplicaset()
+
+		actualRs := nameToReplicasetRef[rs.info.Name]
+		switch {
+		case actualRs == nil:
+			// rs is outdated, can't use it -- let's discover bucket again
+			r.BucketReset(bucketID)
+		case actualRs == rs:
+			return rs, nil
+		default: // actualRs != rs
+			// update rs -> actualRs for this bucket
+			_, _ = r.BucketSet(bucketID, actualRs.info.Name)
+			return actualRs, nil
+		}
 	}
 
 	// it`s ok if in the same time we have few active searches
@@ -106,14 +119,9 @@ func (r *Router) bucketSearchLegacy(ctx context.Context, bucketID uint64) (*Repl
 		return rs, nil
 	}
 
-	/*
-	   -- All replicasets were scanned, but a bucket was not
-	   -- found anywhere, so most likely it does not exist. It
-	   -- can be wrong, if rebalancing is in progress, and a
-	   -- bucket was found to be RECEIVING on one replicaset, and
-	   -- was not found on other replicasets (it was sent during
-	   -- discovery).
-	*/
+	// All replicasets were scanned, but a bucket was not found anywhere, so most likely it does not exist.
+	// It can be wrong, if rebalancing is in progress, and a bucket was found to be RECEIVING on one replicaset,
+	// and was not found on other replicasets (it was sent during discovery).
 
 	return nil, newVShardErrorNoRouteToBucket(bucketID)
 }
@@ -192,6 +200,8 @@ func (r *Router) DiscoveryHandleBuckets(ctx context.Context, rs *Replicaset, buc
 			continue
 		}
 
+		// NOTE: oldRs and rs might have the same name, we intentionally don't check this case to keep the logic simple
+
 		// We don't check oldRs for nil here, because it's a valid key too (if rs == nil, it means removed from unknown buckets set)
 		removedFrom[oldRs]++
 	}
@@ -294,9 +304,15 @@ func (r *Router) cronDiscovery(ctx context.Context) {
 					// Another one panic may happen due to log function below (e.g. bug in log().Errorf), in this case we have two options:
 					// 1. recover again and log nothing: panic will be muted and lost
 					// 2. don't try to recover, we hope that the second panic will be logged somehow by go runtime
-					// So, we choose the second behavior
-					r.log().Errorf(ctx, "[DISCOVERY] something unexpected has happened in cronDiscovery(%d): panic %v, stackstrace: %s",
-						iterationCount, recovered, string(debug.Stack()))
+					// So, we desided to combine them in the third behavior: log in another goroutin
+					iterationCount := iterationCount
+					// get stacktrace in the current goroutine
+					debugStack := string(debug.Stack())
+
+					go func() {
+						r.log().Errorf(ctx, "[DISCOVERY] something unexpected has happened in cronDiscovery(%d): panic %v, stacktrace: %s",
+							iterationCount, recovered, debugStack)
+					}()
 				}
 			}()
 

topology.go

@@ -25,6 +25,14 @@ type TopologyController interface {
 	AddReplicasets(ctx context.Context, replicasets map[ReplicasetInfo][]InstanceInfo) error
 }
 
+func copyMap[K comparable, V any](m map[K]V) map[K]V {
+	copy := make(map[K]V)
+	for k, v := range m {
+		copy[k] = v
+	}
+	return copy
+}
+
 func (r *Router) getNameToReplicaset() map[string]*Replicaset {
 	r.nameToReplicasetMutex.RLock()
 	nameToReplicasetRef := r.nameToReplicaset
@@ -119,10 +127,6 @@ func (r *Router) AddReplicaset(ctx context.Context, rsInfo ReplicasetInfo, insta
 		return ErrReplicasetExists
 	}
 
-	replicaset := &Replicaset{
-		info: rsInfo,
-	}
-
 	rsInstances := make([]pool.Instance, 0, len(instances))
 	for _, instance := range instances {
 		rsInstances = append(rsInstances, pool.Instance{
@@ -158,13 +162,13 @@ func (r *Router) AddReplicaset(ctx context.Context, rsInfo ReplicasetInfo, insta
 		r.log().Errorf(ctx, "got connected now as false to pool.RW")
 	}
 
-	replicaset.conn = conn
+	replicaset := &Replicaset{
+		info: rsInfo,
+		conn: conn,
+	}
 
 	// Create an entirely new map object
-	nameToReplicasetNew := make(map[string]*Replicaset)
-	for k, v := range nameToReplicasetOld {
-		nameToReplicasetNew[k] = v
-	}
+	nameToReplicasetNew := copyMap(nameToReplicasetOld)
 	nameToReplicasetNew[rsInfo.Name] = replicaset // add when conn is ready
 
 	// We could detect concurrent access to the TopologyController interface
@@ -202,10 +206,7 @@ func (r *Router) RemoveReplicaset(ctx context.Context, rsName string) []error {
 	}
 
 	// Create an entirely new map object
-	nameToReplicasetNew := make(map[string]*Replicaset)
-	for k, v := range nameToReplicasetOld {
-		nameToReplicasetNew[k] = v
-	}
+	nameToReplicasetNew := copyMap(nameToReplicasetOld)
 	delete(nameToReplicasetNew, rsName)
 
 	r.setNameToReplicaset(nameToReplicasetNew)

vshard.go

@@ -42,9 +42,8 @@ type Router struct {
 
 	routeMap atomic.Pointer[routeMap]
 
-	// ----------------------- Map-Reduce -----------------------
-	// Storage Ref ID. It must be unique for each ref request
-	// and therefore is global and monotonically growing.
+	// refID is used for Map-Reduce operation. Since it must be unique (within connection) for each ref request,
+	// we made it global and monotonically growing for each Router instance.
 	refID atomic.Int64
 
 	cancelDiscovery func()