fix(executor): re-seat a stale region latch when the pool changes

changelog/unreleased/reseat-stale-region-latch.md

@@ -0,0 +1,7 @@
+### English
+
+- A routing region latch no longer survives a large change in pool composition. Adding or migrating many accounts of another region now re-seats the route on the largest region instead of leaving the new accounts idle.
+
+### 中文
+
+- 路由的区域锁定不再跨越大规模的账号池变化。新增或迁移大量其他区域的账号后，路由会重新落到账号最多的区域，而不会让新账号闲置。

internal/executor/pool.go

@@ -511,8 +511,14 @@ func (p *Pool) PickRoute(q RouteQuery) (Item, bool) {
 	// capture the route.
 	if q.RegionFilter == "" {
 		if previous, ok := p.lastRegion[key]; ok {
-			if subset := inRegion(available, previous); len(subset) > 0 {
+			// The latch is a hint, not a commitment: the pool can change
+			// underneath it. Keep the remembered region only while it still
+			// carries a fair share of the route; a latch left far behind by
+			// a bulk pool change is dropped so the route re-seats below.
+			if subset := inRegion(available, previous); len(subset) > 0 && !regionLatchStale(len(subset), available) {
 				available = subset
+			} else {
+				delete(p.lastRegion, key)
 			}
 		}
 		if _, ok := p.lastRegion[key]; !ok {
@@ -638,6 +644,31 @@ func (p *Pool) ensureRotationKey(key string) {
 	}
 }
 
+// largestRegionCount returns how many candidates sit in the biggest region.
+func largestRegionCount(items []Item) int {
+	counts := map[string]int{}
+	for _, item := range items {
+		counts[itemRegion(item)]++
+	}
+	best := 0
+	for _, count := range counts {
+		if count > best {
+			best = count
+		}
+	}
+	return best
+}
+
+// regionLatchStale reports whether a route's remembered region has been left
+// far behind by the rest of its candidates. The factor-of-two hysteresis keeps
+// a roughly balanced pool on its current region instead of thrashing between
+// two comparable regions, while a latch left behind by a bulk pool change is
+// dropped so the route re-seats on the largest region. Must be called with
+// p.mu held.
+func regionLatchStale(latchedCount int, available []Item) bool {
+	return latchedCount*2 < largestRegionCount(available)
+}
+
 // largestRegion returns the region with the most candidates, breaking ties by
 // name so the choice is stable across restarts rather than map-order random.
 func largestRegion(items []Item) string {

internal/executor/pool_route_test.go

@@ -300,3 +300,60 @@ func TestPickRouteRegionScopedFailoverStaysInsideGrant(t *testing.T) {
 		t.Fatal("exhausted cn grant must not fall through to global")
 	}
 }
+
+// A route latches onto one region so a balanced pool does not flip between
+// regions as rotation advances. The latch must not survive a bulk change in
+// pool composition: landing many accounts of another region on the route has
+// to re-seat it, otherwise the new accounts never receive any traffic.
+func TestPickRouteReseatsStaleRegionLatch(t *testing.T) {
+	p := NewPool(nil, nil)
+	p.Upsert(Item{ID: "g1", URL: "http://g1", Provider: "workbuddy", Region: "global", Runtime: "in_process"})
+	p.Upsert(Item{ID: "g2", URL: "http://g2", Provider: "workbuddy", Region: "global", Runtime: "in_process"})
+
+	for i := 0; i < 5; i++ {
+		got, ok := p.PickRoute(RouteQuery{ProviderFilter: "workbuddy"})
+		if !ok || got.Region != "global" {
+			t.Fatalf("cold route pick %d = %+v ok=%v, want global", i, got, ok)
+		}
+	}
+
+	// A migration lands many CN accounts on the same route.
+	for i := 0; i < 10; i++ {
+		id := "c" + itoa(i)
+		p.Upsert(Item{ID: id, URL: "http://" + id, Provider: "workbuddy", Region: "cn", Runtime: "in_process"})
+	}
+
+	seen := map[string]int{}
+	for i := 0; i < 12; i++ {
+		got, ok := p.PickRoute(RouteQuery{ProviderFilter: "workbuddy"})
+		if !ok {
+			t.Fatalf("pick %d after migration failed", i)
+		}
+		seen[got.Region]++
+	}
+	if seen["cn"] == 0 {
+		t.Fatalf("stale global latch starved the new cn accounts: %v", seen)
+	}
+	if seen["global"] != 0 {
+		t.Fatalf("route must re-seat fully on the largest region: %v", seen)
+	}
+}
+
+// The hysteresis must not re-seat a roughly balanced pool, or the route would
+// thrash between two comparable regions.
+func TestPickRouteKeepsRegionLatchWhenBalanced(t *testing.T) {
+	p := NewPool(nil, nil)
+	p.Upsert(Item{ID: "g1", URL: "http://g1", Provider: "workbuddy", Region: "global", Runtime: "in_process"})
+	p.Upsert(Item{ID: "c1", URL: "http://c1", Provider: "workbuddy", Region: "cn", Runtime: "in_process"})
+
+	first, ok := p.PickRoute(RouteQuery{ProviderFilter: "workbuddy"})
+	if !ok {
+		t.Fatal("first pick failed")
+	}
+	for i := 0; i < 8; i++ {
+		got, ok := p.PickRoute(RouteQuery{ProviderFilter: "workbuddy"})
+		if !ok || got.Region != first.Region {
+			t.Fatalf("balanced pool flipped region: first=%s pick %d = %+v ok=%v", first.Region, i, got, ok)
+		}
+	}
+}