Exclude scheduled-to-stop members from resolver availability

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,38 @@ 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 but are not available for new requests.
+	s.Len(resolver.Members(), 3)
+	s.Equal(3, resolver.MemberCount())
+	s.Len(resolver.AvailableMembers(), 1)
+	s.Equal(1, resolver.AvailableMemberCount())
+	s.Equal(running.GetAddress(), resolver.AvailableMembers()[0].GetAddress())
+
+	// But they remain in the lookup ring, so ownership transfers only at eviction.
+	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,16 @@ func isDraining(host *hostInfo) bool {
 	}
 	return false
 }
+
+func hasStopAt(host *hostInfo) bool {
+	_, ok := host.Label(stopAtKey)
+	return ok
+}
+
+// isLeaving reports that a host has announced its departure (draining or a
+// scheduled stop). Such hosts are dropped from AvailableMembers and
+// AvailableMemberCount, but stay in the lookup ring until they actually leave
+// so shard/task-queue ownership transfers only at eviction.
+func isLeaving(host *hostInfo) bool {
+	return isDraining(host) || hasStopAt(host)
+}