Exclude scheduled-to-stop members from resolver availability

AvailableMembers/AvailableMemberCount already excluded draining hosts;
extend that to hosts with a future stopAt label. Such a host has
announced its departure and flips itself to NOT_SERVING, so it should
not be offered for new request selection or counted as available, the
same way a draining frontend isn't.

The consistent-hash lookup ring is intentionally left untouched: a
leaving host stays in the ring until its scheduled eviction so that
shard/task-queue ownership transfers once, simultaneously across nodes,
rather than diverging during the drain window.

Author: prathyushpv

common/membership/interfaces.go

@@ -80,11 +80,13 @@ type (
 		RemoveListener(name string) error
 		// MemberCount returns the number of known hosts running this service.
 		MemberCount() int
-		// AvailableMemberCount returns the number of hosts running this service that are accepting requests (not draining).
+		// AvailableMemberCount returns the number of hosts running this service that are accepting
+		// requests (not draining and not scheduled to stop).
 		AvailableMemberCount() int
 		// Members returns all known hosts available for this service.
 		Members() []HostInfo
-		// AvailableMembers returns all hosts available for this service that are accepting requests (not draining).
+		// AvailableMembers returns all hosts available for this service that are accepting requests
+		// (not draining and not scheduled to stop).
 		AvailableMembers() []HostInfo
 		// RequestRefresh requests that the membership information be refreshed.
 		RequestRefresh()

common/membership/ringpop/monitor_test.go

@@ -222,6 +222,42 @@ func (s *RpoSuite) TestCompareMembersWithDraining() {
 	s.Len(resolver.AvailableMembers(), 2)
 }
 
+func (s *RpoSuite) TestAvailableMembersExcludeLeaving() {
+	running := newHostInfo("a", nil)
+	draining := newHostInfo("b", map[string]string{drainingKey: "true"})
+	stopping := newHostInfo("c", map[string]string{stopAtKey: "9999999999"})
+	hosts := []*hostInfo{running, draining, stopping}
+
+	ring := newHashRing(100)
+	for _, h := range hosts {
+		ring.AddMembers(h)
+	}
+	resolver := &serviceResolver{}
+	resolver.ringAndHosts.Store(ringAndHosts{
+		ring: ring,
+		hosts: map[string]*hostInfo{
+			running.GetAddress():  running,
+			draining.GetAddress(): draining,
+			stopping.GetAddress(): stopping,
+		},
+	})
+
+	// Leaving hosts stay visible in full membership.
+	s.Len(resolver.Members(), 3)
+	s.Equal(3, resolver.MemberCount())
+
+	// ...but are not offered as available for new request selection.
+	s.Len(resolver.AvailableMembers(), 1)
+	s.Equal(1, resolver.AvailableMemberCount())
+	s.Equal(running.GetAddress(), resolver.AvailableMembers()[0].GetAddress())
+
+	// Ownership guard: leaving hosts remain in the lookup ring so that
+	// shard/task-queue ownership transfers only once, at their scheduled
+	// eviction, rather than diverging across nodes during the drain window.
+	addrs := ring.LookupN("some-key", 3)
+	s.ElementsMatch([]string{"a", "b", "c"}, addrs)
+}
+
 func eventToString(event *membership.ChangedEvent) []string {
 	var diff []string
 	for _, a := range event.HostsAdded {

common/membership/ringpop/service_resolver.go

@@ -216,7 +216,7 @@ func (r *serviceResolver) AvailableMemberCount() int {
 	_, hosts := r.ring()
 	n := 0
 	for _, host := range hosts {
-		if !isDraining(host) {
+		if !isLeaving(host) {
 			n++
 		}
 	}
@@ -236,7 +236,7 @@ func (r *serviceResolver) AvailableMembers() []membership.HostInfo {
 	_, hosts := r.ring()
 	var servers []membership.HostInfo
 	for _, host := range hosts {
-		if !isDraining(host) {
+		if !isLeaving(host) {
 			servers = append(servers, host)
 		}
 	}
@@ -531,3 +531,20 @@ func isDraining(host *hostInfo) bool {
 	}
 	return false
 }
+
+// hasStopAt reports whether the host has announced a scheduled eviction time.
+// getReachableMembers drops hosts whose stopAt is already in the past, so a
+// label present here always denotes a future eviction.
+func hasStopAt(host *hostInfo) bool {
+	_, ok := host.Label(stopAtKey)
+	return ok
+}
+
+// isLeaving reports that a host has announced it is going away, either by
+// draining or by scheduling a future eviction. Such a host stays in the hash
+// ring until it actually leaves (so shard/task-queue ownership transfers only
+// once, at the scheduled time), but it should no longer be offered for new
+// request selection or counted as available.
+func isLeaving(host *hostInfo) bool {
+	return isDraining(host) || hasStopAt(host)
+}