clientv3: only update initReq.rev == 0 with watch revision

[heyitsanthony]

Always updating the initReq.rev on watch create will resume from the wrong
revision if initReq is ever nonzero.

[xiang90]

add some comments here? (when resume for a disconnection, we start watcher from a known previous revision not from the current revision, so do not update it to header.revision)

[xiang90]

@heyitsanthony How did you find this bug? from a test failure?

[heyitsanthony]

@xiang90 from investigating #7709, but likely an unrelated issue.

[xiang90]

is it possible to develop a dedicated test for the case we fix?

i feel TestWatchResumeCompacted is already hard enough to reason about (it is not deterministic. we have to sub case to care about internally)

[heyitsanthony]

It's non-deterministic by nature since it relies on disconnect behavior. If it's going to be deterministic I'd need to add failpoints to the testing infrastructure. If I wrote a separate test it'd be essentially duplicate what's there now.

[heyitsanthony]

I can try some ways to do deterministic without failpoints but it'll change the behavior that's being tested. OK, whatever.

[xiang90]

The fix looks good to me.

[fanminshi]

This pr fixes my issue as well.

My issue:

1. send create req on initReq{key:foo, rev:2} and recv create resp back.
2. update initReq.rev to resp.Header.Revision let's say it is 100.
3. conn disconnect.
4. watchGrpcStream resends initReq{key:foo, rev:100}.
5. wch hangs because watch on rev 100 instead of rev 2.

It would be nice to have a simple test case that cover the above situation.
I can write one in a different pr when this gets merged.

[fanminshi]

is it possible that before you DropConnections, watch receives and buffers event ("a", rev=3)?

[heyitsanthony]

yes, but I tested it without the fix and it fails reliably.

[fanminshi]

Would the following code fail due to having events buffered?

clus.Members[0].DropConnections()
clus.Members[0].PauseConnections()

select {
 	case resp, ok := <-wch:
 		t.Fatalf("wch should block, got (%+v, %v)", resp, ok)
 	case <-time.After(100 * time.Millisecond):
 	}

[heyitsanthony]

yes, I can make it skip instead of fatal, but CI is typically slower rather than faster

[heyitsanthony]

@fanminshi the current test essentially already does that, all that would have to change is:

if _, err := cli.Put(context.TODO(), "a", "4"); err != nil {

instead having:

if _, err := cli.Put(context.TODO(), "b", "4"); err != nil {

Testing based on a hang won't fail fast-- it'll have to wait for the timeout to catch the failure, instead of failing as soon as it retrieving a bad value. I don't understand how a second test would catch any new failure modes since if there's a timeout for some other reason, it'll still fail. Seems redundant.

[fanminshi]

@heyitsanthony understood!

[xiang90]

@heyitsanthony LGTM. Thanks!

[fanminshi]

lgtm, thanks!