Fixed unexpected cancelled watch with WatchID=0.

[kafuu-chino]

etcdserver: Fixed unexpected cancelled watch with WatchID=0.

I ran into the same problem from #12385, I found that the error will cause the watch with WatchID=0 to be invalid，so I set WatchID to start at 1 to fix it.

First of all WatchID=0 is semantic, it's AutoWatchID.

Secondly when token expired, make a new watch request to server side, serverWatchStream.isWatchPermitted returns false, then send WatchResponse to serverWatchStream.ctrlStream with Canceled=true, however WatchResponse.WatchID=creq.WatchId, default creq.WatchId=0, so a watch with WatchID=0 will be unexpectedly cancelled on the server side and client can never receive messages with WatchID=0.

if !sws.isWatchPermitted(creq) {
    wr := &pb.WatchResponse{
        Header:       sws.newResponseHeader(sws.watchStream.Rev()),
        WatchId:      creq.WatchId, // default to 0
        Canceled:     true, // will delete 0 from ids
        Created:      true,
        CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
    }

    select {
        case sws.ctrlStream <- wr: // send to ctrlStream
            continue
        case <-sws.closec:
            return nil
    }
}

...

case c, ok := <-sws.ctrlStream: // receive from ctrlStream
    if !ok {
        return
    }

    ...

    // track id creation
    wid := mvcc.WatchID(c.WatchId)
    if c.Canceled {
        delete(ids, wid) // delete from ids and client can never receive messages
        continue
    }

[kafuu-chino]

I found some comments that said watchID starts at 0, I don't know if there are any other effects.

[ahrtr]

Thanks @kafuu-chino for the PR, but I did not get what's the issue you are fixing. The PR is also causing some test failures.

[kafuu-chino]

@ahrtr
Let me explain in detail:

1. First, WatchCreateRequest does not set the value of WatchID, so WatchID = 0.

func (wr *watchRequest) toPB() *pb.WatchRequest {
    req := &pb.WatchCreateRequest{
        StartRevision:  wr.rev,
        Key:            []byte(wr.key),
        RangeEnd:       []byte(wr.end),
        ProgressNotify: wr.progressNotify,
        Filters:        wr.filters,
        PrevKv:         wr.prevKV,
        Fragment:       wr.fragment,
    }
    cr := &pb.WatchRequest_CreateRequest{CreateRequest: req}
    return &pb.WatchRequest{RequestUnion: cr}
}

2. On server side, WatchID = 0 means the WatchID is automatically generated, see AutoWatchID.

// AutoWatchID is the watcher ID passed in WatchStream.Watch when no
// user-provided ID is available. If pass, an ID will automatically be assigned.
const AutoWatchID WatchID = 0

Due to ws.nextID = 0, so the first watch with WatchID = 0.

if id == AutoWatchID {
    for ws.watchers[ws.nextID] != nil {
        ws.nextID++
    }
    id = ws.nextID
    ws.nextID++
}

3. Watch once, initialize watch with WatchID = 0 and wait for the token to expire. Then watch again, if the token expires, sws.isWatchPermitted returns false, creq.WatchId = 0, Canceled = true and Created = true at this time, then send WatchResponse to sws.ctrlStream.

if !sws.isWatchPermitted(creq) {
    wr := &pb.WatchResponse{
        Header:       sws.newResponseHeader(sws.watchStream.Rev()),
        WatchId:      creq.WatchId, // = 0
        Canceled:     true,
        Created:      true,
        CancelReason: rpctypes.ErrGRPCPermissionDenied.Error(),
    }

    select {
        case sws.ctrlStream <- wr:
            continue
        case <-sws.closec:
            return nil
    }
}

4. Next handle WatchResponse on after Send, if Canceled == true, then delete(ids, wid), wid = 0 at this time. This operation causes the server considers the watch with WatchID = 0 to be no longer active.

case c, ok := <-sws.ctrlStream:
    if !ok {
        return
    }

if err := sws.gRPCStream.Send(c); err != nil {
    if isClientCtxErr(sws.gRPCStream.Context().Err(), err) {
        sws.lg.Debug("failed to send watch control response to gRPC stream", zap.Error(err))
    } else {
        ws.lg.Warn("failed to send watch control response to gRPC stream", zap.Error(err))
        streamFailures.WithLabelValues("send", "watch").Inc()
    }
    return
}

// track id creation
wid := mvcc.WatchID(c.WatchId)
if c.Canceled {
    delete(ids, wid)
    continue
}

5. Client side receives WatchResponse, then enter switch case pbresp.Created:. The new watch returns an error, the old watch with WatchID = 0 does not respond, and no longer receives messages from the server. This is not what we want, we just want this watch to fail, and we don't want to make the previous watch unresponsive on the server side.

case pbresp := <-w.respc:
    if cur == nil || pbresp.Created || pbresp.Canceled {
        cur = pbresp
    } else if cur != nil && cur.WatchId == pbresp.WatchId {
    // merge new events
    cur.Events = append(cur.Events, pbresp.Events...)
    // update "Fragment" field; last response with "Fragment" == false
    cur.Fragment = pbresp.Fragment
    }

    switch {
    case pbresp.Created:

[kafuu-chino]

I looked at the commit history and restored creq.WatchId to -1.

[ahrtr]

Probably we should check the value of creq.WatchId, how about the logic something like below?

watchId := creq.WatchId
if watchId == int64(mvcc.AutoWatchID) {
	watchId = -1
}

Another choice is to make watchId starting with 1 instead of 0. But we need to fix some test cases, and it also change the default behavior, accordingly might have some impact on client applications which depend on the watch.

I manually verified that this PR indeed can fix the issue #12385.

But the overall design looks weird to me. The existing watch can continue to work even the simple token expired, because the client can retry the authentication. But creating new watch stream using exactly the same client doesn't work, it's a little strange.

It seems that we need to get the overall design clarified firstly. WDYT @mitake

[kafuu-chino]

I set watchId starting with 1 at the beginning, all issues are resolved and do not conflict with AutoWatchID but test failed. If can fix test, I think it is better to start with 1.

I haven't found a place to set creq.WatchId yet, so it should always be mvcc.autowatchid?

[mitake]

I think it's better to use -1 as an invalid watch ID for preserving the current behavior of watch API like what current PR does. How about introducing a new const like AutoWatchID?

Let me check the behavior introduced by this PR in a few days. I only tested a case which don't rely on it.

[ahrtr]

I haven't found a place to set creq.WatchId yet, so it should always be mvcc.autowatchid?

Yes, it's basically true. In theory, users do not necessarily use the etcd client SDK. When they construct the wathRequest themselves, then watchId might not be 0 anymore. But the possibility should be very low, and it isn't the recommended way. So probably we can assume that it's always mvcc.AutoWatchID for any pb.WatchRequest_CreateRequest.

I think it's better to use -1 as an invalid watch ID for preserving the current behavior of watch API like what current PR does.

Based on point above (watchId always be mvcc.AutoWatchID for any pb.WatchRequest_CreateRequest), it makes sense to always use -1 as an invalid watch ID

[mitake]

In theory, users do not necessarily use the etcd client SDK. When they construct the wathRequest themselves, then watchId might not be 0 anymore. But the possibility should be very low, and it isn't the recommended way.

Yeah using non standard SDK is technically possible... but I think it's good to exclude it because it's quite irregular as you point out.

[kafuu-chino]

@mitake
I only found two places to return both Canceled and Created are true and can trigger the assert.

1. https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L271
2. https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L308

What do you think which cause assert triggerred? Or i missed something and can you help to supplement?

I agree with :

Passing InvalidWatchID to delete() or looking it up from ids isn't harmful.

Because it will not affect the normal watch.

Does c.CancelReason == "" equal to c.Canceled == false? I think it means the watch will not be deleted and the WatchId will be a normal watch id.

In my opinion, the purpose of this assert is that when a watch is newly created and an error is encountered ( both Canceled and Created are true), do not cause other normal watches to be deleted (WatchId MUST BE InvalidWatchID), it's hard to detect when it happens.

[mitake]

@kafuu-chino In the case of TestKVCompaction, https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L308 is causing the assert. The condition is c.Created == true && c.Canceled == false && c.WatchId == 0 && c.CancelReason == "".

On the second thought, the condition should be like (c.Canceled || c.Created) && wid != mvcc.InvalidWatchID? If it's false, the branches https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L468-L472 will be executed and unintentional operations against invalid watch ID will happen (although it's not harmful).

It's great if you can try to run the test case too: TESTCASE=TestKVCompact TEST_OPTS="PASSES='integration'" make test

[kafuu-chino]

@mitake
The assert condition really wrong and I've got the condition backwards.

If condition like (c.Canceled || c.Created) && wid != mvcc.InvalidWatchID, there will be problems here: https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L342

1. c.Created == false and c.Canceled == false
2. then (c.Canceled || c.Created) == false
3. and (c.Canceled || c.Created) && wid != mvcc.InvalidWatchID == false
4. condition == false and assert will be trigger

How about !(c.Canceled && c.Created) || wid == mvcc.InvalidWatchID？

[mitake]

@kafuu-chino Sorry the above my condition is very wrong... Especially if it includes && wid != mvcc.InvalidWatchID, it cannot handle an invalid wid always. I guess I pasted a wrong version :(
I think your condition is correct. Sorry for the confusion.

[mitake]

Let me review the entire PR, I think I can reply sometime tomorrow.

[kafuu-chino]

@ahrtr @mitake
Help review it sometime.

[mitake]

@kafuu-chino Could you squash the 2 commits into one, and change the commit title like this? *: avoid closing a watch with ID 0 incorrectly
I think the test failure isn’t related to your change, so running tests on my local machine. TestKVCompact is failing because of the assert.

[kafuu-chino]

@mitake Sure, I've squashed and rebase to the latest commit.

[ahrtr]

I think WatchResponse with Created && Canceled and valid ID is possible in other paths, e.g. https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L308
Could you update the assert condition for checking CancelReason? @kafuu-chino

The Watch always returns -1 when there is an error. Please replace all the -1 with the const InvalidWatchID. cc @kafuu-chino

[ahrtr]

@kafuu-chino The change looks good to me. But please rebase this PR instead of merging main into your dev branch.
Firstly sync your main branch with the upstream etcd. Afterwards, execute commands something like below,

$ git checkout main
$ git pull
$ git checkout fix
$ git rebase -i main  # keep the "pick" in the first line, and change all the following lines' "pick" to "s"

[kafuu-chino]

@ahrtr
Like this?

[mitake]

Overall LGTM, could you rebase on the latest main branch correctly? The commit 9fba81a comes from the PR #14377 so it shouldn't be included in this PR.

server/etcdserver/api/v3rpc/watch.go, especially serverWatchStream.recvLoop() is hard to test with unit tests. Do you think it's possible to add integration or e2e test?

[ahrtr]

Please update per @mitake 's comment.

You need to rebase this PR instead of merging others branch into it.

[kafuu-chino]

@mitake
That's the idea for my local test.

1. Watch a key.
2. Watch a new key and enter this code https://github.com/etcd-io/etcd/blob/main/server/etcdserver/api/v3rpc/watch.go#L271. I set the --auth-token-ttl of the server and wait for the token to expire.
3. Set new value of key in step 1, and check whether message from watch can be received normally.

I need to read the tests package code first and it's take some time to think test code.

[mitake]

@kafuu-chino I'm writing a test case in ff9c661 for #14322
For now the test case is still in progress (for some reasons the unary interceptor refreshes the token so the mechanism I'm adding isn't used), but it might also be useful for this PR too. If you want to try it as a starting point, feel free to use it.

[ahrtr]

LGTM

Thank you @kafuu-chino

The only concern is that there is no new test case for the code change. But it's OK if @mitake or @kafuu-chino can address the test case in a separate PR.

Leave it to @mitake to approve & merge this PR.

[mitake]

@kafuu-chino @ahrtr I finalized the test for my PR in 94fd161 I think it can be extended for covering the change in this PR. Let me think about it. Probably I'll be able to share my update sometime early next week.

[mitake]

@kafuu-chino On the second thought, I think this change is essentially not limited to auth token TTL. I wrote below integration test and checked that the test can fail in the main branch and run successfully with your branch. Could you check it's a suitable test case for your change? If it's ok, could you include it in this PR? (path is tests/integration/v3_auth_test.go)

func TestV3AuthWatchErrorAndWatchId0(t *testing.T) {
	integration.BeforeTest(t)
	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)

	ctx, cancel := context.WithTimeout(context.TODO(), 10*time.Second)
	defer cancel()

	users := []user{
		{
			name:     "user1",
			password: "user1-123",
			role:     "role1",
			key:      "k1",
			end:      "k2",
		},
	}
	authSetupUsers(t, integration.ToGRPC(clus.Client(0)).Auth, users)

	authSetupRoot(t, integration.ToGRPC(clus.Client(0)).Auth)

	c, cerr := integration.NewClient(t, clientv3.Config{Endpoints: clus.Client(0).Endpoints(), Username: "user1", Password: "user1-123"})
	if cerr != nil {
		t.Fatal(cerr)
	}
	defer c.Close()

	watchEndCh := make(chan interface{})

	go func() {
		wChan := c.Watch(ctx, "k1", clientv3.WithRev(1))
		watchResponse := <-wChan
		fmt.Printf("watch response from k1: %v\n", watchResponse)
		testutil.AssertTrue(t, len(watchResponse.Events) != 0)
		watchEndCh <- struct{}{}
	}()

	// Sleep for making sure that the above goroutine invokes Watch()
	// So the above Watch() can get watch ID = 0
	time.Sleep(1 * time.Second)

	wChan := c.Watch(ctx, "non-allowed-key", clientv3.WithRev(1))
	watchResponse := <-wChan
	testutil.AssertNotNil(t, watchResponse.Err()) // permission denied

	_, err := c.Put(ctx, "k1", "val")
	if err != nil {
		t.Fatalf("Unexpected error from Put: %v", err)
	}

	<-watchEndCh
}

[kafuu-chino]

@mitake Ok, I'll try the test later.

[kafuu-chino]

@mitake
I think the test is fine and add it to the PR.

[mitake]

@kafuu-chino Could you rebase this PR based on the latest main? My PR introduced a conflict in the integration test file, sorry.

[kafuu-chino]

@mitake
Rebase finished.

[mitake]

LGTM, thanks a lot @kafuu-chino ! It's great if @ahrtr can cross check the test case.

[ahrtr]

The test case is OK, but I do not any relationship between this case and the issue this PR is trying to fix.

[mitake]

Actually auth token TTL expiration isn't the only one reason which can cause unexpected cancelling watch with ID=0. Other errors caused by isWatchPermitted() https://github.com/etcd-io/etcd/pull/14296/files#diff-79ca72cbf269ee36941406748b6078905b26447428c7179f70e543669cf42ec3R272 can result the same issue. So I used the different error auth.ErrPermissionDenied for checking the corrected behavior. By using this error, the test case can avoid sleeping for waiting auth token TTL and it's better for making it deterministically.

[ahrtr]

Thanks @mitake for the explanation.

Actually auth token TTL expiration isn't the only one reason which can cause unexpected cancelling watch with ID=0.

I follow this issue intermittently, so I might lose some context. I agree with you on your point, but I don't think it verifies the scenario of the original issue #12385 ? My understand is that the client should can successfully Watch even when the auth token is expired, so we need to add a case to verify it. Please correct me if I am wrong.

[mitake]

My understand is that the client should can successfully Watch even when the auth token is expired, so we need to add a case to verify it.

Correct, and this behavior is checked by the test case added in another PR: #14322 The test case is https://github.com/etcd-io/etcd/pull/14322/files#diff-c42aac07498140d89b902c1e02edb73076615172896ba431ec99319636432e1dR501 (TestV3AuthWatchAndTokenExpire). So the test case added in this PR (TestV3AuthWatchErrorAndWatchId0) focuses only on the issue happens to watch with ID=0. I think the 2 test cases are enough for covering the issues reported in #12385 (actually there are 2 related but independent issues). How do you think @ahrtr ?

[ahrtr]

Thanks @mitake for the clarification, which makes sense.

[kafuu-chino]

@mitake @ahrtr modification of test has committed, please review.

[codecov-commenter]

Codecov Report

Merging #14296 (f1d4935) into main (434c7c4) will decrease coverage by 0.10%.
The diff coverage is 100.00%.

@@            Coverage Diff             @@
##             main   #14296      +/-   ##
==========================================
- Coverage   75.35%   75.24%   -0.11%     
==========================================
  Files         457      457              
  Lines       37190    37193       +3     
==========================================
- Hits        28023    27986      -37     
- Misses       7408     7442      +34     
- Partials     1759     1765       +6     

Flag	Coverage Δ
all	75.24% <100.00%> (-0.11%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Impacted Files	Coverage Δ
client/v3/watch.go	91.81% <100.00%> (-0.59%)	⬇️
server/etcdserver/api/v3rpc/watch.go	86.03% <100.00%> (+0.45%)	⬆️
server/proxy/grpcproxy/watch.go	92.48% <100.00%> (-1.16%)	⬇️
server/storage/mvcc/watcher.go	100.00% <100.00%> (ø)
client/pkg/v3/tlsutil/tlsutil.go	83.33% <0.00%> (-8.34%)	⬇️
server/storage/mvcc/watchable_store.go	84.42% <0.00%> (-7.98%)	⬇️
client/v3/experimental/recipes/queue.go	58.62% <0.00%> (-6.90%)	⬇️
client/v3/leasing/util.go	95.00% <0.00%> (-3.34%)	⬇️
server/etcdserver/api/v3rpc/util.go	70.96% <0.00%> (-3.23%)	⬇️
server/etcdserver/api/v3rpc/member.go	93.54% <0.00%> (-3.23%)	⬇️
... and 16 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[ahrtr]

Overall look good to me. Thanks!

[mitake]

@kafuu-chino Thanks a lot for working on this! Is it possible for you to open PRs which backport this change to 3.4 and 3.5 branches?

[mitake]

@kafuu-chino please let me know if you don’t have time to backport the change to release-3.4 and 3.5. Probably I’ll be able to help.

[kafuu-chino]

@kafuu-chino please let me know if you don’t have time to backport the change to release-3.4 and 3.5. Probably I’ll be able to help.

Sorry, I just got back from my vacation, I'll handle it today.