test(k8sresolver): assert on convergence, not the first update (#1002)

Fixes #1001

`TestK8sResolverEndpointSliceUpdates` intermittently fails at
`resolver_test.go:215` — 8 of the 30 most recent failed `pr-workflow`
runs (~27%):

```
updated state.Addresses = [{Addr: "10.0.0.1:443", ServerName: "", }],
                    want [{Addr: "10.0.0.1:443", ServerName: "", } {Addr: "10.0.0.2:443", ServerName: "", }]
```

`Build` starts a goroutine that calls `updateState` once
`WaitForCacheSync` returns. That report is deliberate: a service with no
EndpointSlices never fires `AddFunc`, so without it the resolver would
stay silent instead of telling gRPC the answer is an empty set. But
`WaitForCacheSync` polls at `syncedPollPeriod = 100ms`, so it fires
roughly 100ms after Build — and if the test has not yet created the
second slice by then, that update still carries only `10.0.0.1` and sits
in the channel ahead of the real one. The second `select` took whatever
came next, so it asserted against the stale update.

Locally Build-to-Create is 0.6ms, well ahead of the timer, which is why
this only shows up on loaded runners.

The resolver is not at fault — it promises eventual convergence, not
that the first update after a change is final, and a duplicate update
costs gRPC nothing. So both waits now go through one `waitForAddrs`
helper that consumes updates until the set matches, with a timeout so a
genuinely broken resolver still fails and reports the last set it saw.
Note this changes the first wait as well: it asserted the *first* update
equals `[10.0.0.1]`, and now waits for that set instead. The same
argument applies there — nothing promises the first update is final.

## Verification

`-count=N` proves nothing here: the unfixed test passes locally at any
count because the window is never hit. A/B with the Build→Create delay
as the only variable:

| delay | old assertion | new assertion |
|---|---|---|
| 0ms | 5/5 pass | 5/5 pass |
| 150ms | **0/5 pass** | **5/5 pass** |

Measured timeline with the 150ms stall in place:

```
[  0.5ms]  update #1: [10.0.0.1]              <- AddFunc for slice1
           first select takes it
           ... 150ms stall ...
[101.1ms]  update #2: [10.0.0.1]              <- the WaitForCacheSync goroutine
           test creates slice2
[151.4ms]  update #3: [10.0.0.1, 10.0.0.2]
```

The probe tests used for this are not included.

Rebased over #1013. That fixes a different bug — concurrent
`updateState` calls letting an older address set win — and does not
close this one: the two updates here are ~100ms apart, so the queue has
nothing to coalesce. Re-measured on top of it, unchanged: at a 150ms
delay the old assertion is 0/5 and the new one 5/5.

internal/k8sresolver/resolver_test.go

@@ -54,6 +54,25 @@ func (c *testClientConn) ParseServiceConfig(serviceConfigJSON string) *serviceco
 	return nil
 }
 
+// waitForAddrs consumes updates until the set matches want: the resolver promises
+// eventual convergence, not that the first update after a change is final.
+func waitForAddrs(t *testing.T, cc *testClientConn, want []resolver.Address) {
+	t.Helper()
+	deadline := time.After(2 * time.Second)
+	var last []resolver.Address
+	for {
+		select {
+		case state := <-cc.stateChan:
+			last = state.Addresses
+			if reflect.DeepEqual(last, want) {
+				return
+			}
+		case <-deadline:
+			t.Fatalf("state.Addresses never converged: last = %v, want %v", last, want)
+		}
+	}
+}
+
 func TestParseTarget(t *testing.T) {
 	tests := []struct {
 		name          string
@@ -168,15 +187,7 @@ func TestK8sResolverEndpointSliceUpdates(t *testing.T) {
 	}
 	defer res.Close()
 
-	select {
-	case state := <-cc.stateChan:
-		wantAddrs := []resolver.Address{{Addr: "10.0.0.1:443"}}
-		if !reflect.DeepEqual(state.Addresses, wantAddrs) {
-			t.Errorf("initial state.Addresses = %v, want %v", state.Addresses, wantAddrs)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for initial state update")
-	}
+	waitForAddrs(t, cc, []resolver.Address{{Addr: "10.0.0.1:443"}})
 
 	// Add a new EndpointSlice
 	slice2 := &discoveryv1.EndpointSlice{
@@ -205,18 +216,10 @@ func TestK8sResolverEndpointSliceUpdates(t *testing.T) {
 		t.Fatalf("failed to create slice2: %v", err)
 	}
 
-	select {
-	case state := <-cc.stateChan:
-		wantAddrs := []resolver.Address{
-			{Addr: "10.0.0.1:443"},
-			{Addr: "10.0.0.2:443"},
-		}
-		if !reflect.DeepEqual(state.Addresses, wantAddrs) {
-			t.Errorf("updated state.Addresses = %v, want %v", state.Addresses, wantAddrs)
-		}
-	case <-time.After(2 * time.Second):
-		t.Fatal("timed out waiting for state update after slice addition")
-	}
+	waitForAddrs(t, cc, []resolver.Address{
+		{Addr: "10.0.0.1:443"},
+		{Addr: "10.0.0.2:443"},
+	})
 }
 
 func TestK8sResolverClose(t *testing.T) {