Filed flaky test report #17556, provided root cause and how to fix.

Is there any way to re-test this specific failed test E2E / test (linux-386)? @jmhbnz

/retest

(Will only retest workflows that have failed by default)

Thanks for the fix.

The fix is simple and makes sense to me, but the 230+ lines of e2e test is a little over complicated to me. Are you able to create a simple unit test to simulate the case that w.ch is full instead?

Also when I read the source code, I found there maybe another slightly related issue. As the code showed above, when w.minRev < compactRev, etcd will send out a WatchResponse with CompactRevision, and also remove the watcher wg.delete(w), but when the watcher count > maxWatchersPerSync (512), it only removes the watcher from a temporary watcherGroup (see below). So it doesn't really remove the watcher? Could you double confirm this, e.g by writing a unit test? Thanks.

@ahrtr @serathius

it only removes the watcher from a temporary watcherGroup (see below). So it doesn't really remove the watcher? Could you double confirm this, e.g by writing a unit test? Thanks.

Because the .choose function is written badly. It sometimes mutates the original watchGroup, sometimes returns the same watch group, sometimes a copy. It's not great.

Yeah, I have confirmed it would cause compacted watcher still exists in the original watcher group after wg.choose with the following unit test.

Also I observed that slow watchers metrics is 829 even if I only opened 801 watchers. Haven't yet dig into that but it could be related.

func TestWatchGroupUpdate(t *testing.T) {
	ch := make(chan WatchResponse, chanBufLen)
	compactRev := int64(5)
	curRev := int64(12)
	wg := newWatcherGroup()
	wg.add(&watcher{
		key:    []byte("foo/"),
		end:    []byte("foo0"),
		minRev: 2,
		id:     0,
		ch:     ch,
	})
	wg.add(&watcher{
		key:    []byte("foo/"),
		end:    []byte("foo0"),
		minRev: 3,
		id:     1,
		ch:     ch,
	})

	wg.choose(1 /* maxWatchers */, curRev, compactRev)
	// we would expect whatever the picked compacted watcher should be deleted from the watch group
	for w := range wg.watchers {
		t.Logf("compactRev is: %d; watcher %d with minRev %d still in the watcher group", compactRev, w.id, w.minRev)
	}
	require.Equal(t, 1, len(wg.watchers))
}

dev-dsk-chaochn-2c-a26acd76 % GOWORK=off go test -v -run TestWatchGroupUpdate
=== RUN   TestWatchGroupUpdate
    watcher_group_test.go:32: compactRev is: 5; watcher 0 with minRev 2 still in the watcher group
    watcher_group_test.go:32: compactRev is: 5; watcher 1 with minRev 3 still in the watcher group
    watcher_group_test.go:34:
        	Error Trace:	/home/chaochn/workplace/EKS-etcd/src/EKS-etcd/server/storage/mvcc/watcher_group_test.go:34
        	Error:      	Not equal:
        	            	expected: 1
        	            	actual  : 2
        	Test:       	TestWatchGroupUpdate
--- FAIL: TestWatchGroupUpdate (0.00s)
FAIL
exit status 1
FAIL	go.etcd.io/etcd/server/v3/storage/mvcc	0.011s

I can submit another separate PR to fix it.

Are you able to create a simple unit test to simulate the case that w.ch is full instead?

I would like to keep the existing e2e test to ensure no regression of etcd watch behavior after the fix gets in. I would simplify it by removing unnecessary code paths.

Unit test is good to have and I am working on it.

Please ping me back when you resolve all comments.

I would like to keep the existing e2e test to ensure no regression of etcd watch behavior after the fix gets in. I would simplify it by removing unnecessary code paths.

Unit test is good to have and I am working on it.

OK

I think we might need to do more changes in the watch code soon.

Let's go with a minimal fix proposed by the original PR to address the issue.

@chaochn47 can you simplify the e2e test? I should be able to help with if needed.

Ping @ahrtr @serathius

Have you considered using a gofailpoint to simulate watch stream being clogged?

I managed to reproduce the issue using a following test:

func TestV3NoEventsLostOnCompact(t *testing.T) {
	if integration.ThroughProxy {
		t.Skip("grpc proxy currently does not support requesting progress notifications")
	}
	integration.BeforeTest(t)

	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)

	client := clus.RandClient()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	writeCount := mvcc.WatchStreamResponseBufferLen * 11 / 10

	wch := client.Watch(ctx, "foo")
	require.NoError(t, gofail.Enable("watchResponseSend", `sleep(1000)`))
	var rev int64 = 0
	for i := 0; i < writeCount; i++ {
		resp, err := client.Put(ctx, "foo", "bar")
		require.NoError(t, err)
		rev = resp.Header.Revision
	}
	_, err := client.Compact(ctx, rev)
	require.NoError(t, err)
	time.Sleep(time.Second)
	require.NoError(t, gofail.Disable("watchResponseSend"))

	event_count := 0
	compacted := false
	for resp := range wch {
		err = resp.Err()
		if err != nil {
			if !strings.Contains(err.Error(), "required revision has been compacted") {
				t.Fatal(err)
			}
			compacted = true
			break
		}
		event_count += len(resp.Events)
		if event_count == writeCount {
			break
		}
	}
	assert.Truef(t, compacted, "Expected stream to get compacted, instead we got %d events out of %d events", event_count, writeCount)
}

Where watchResponseSend is a new failpoint before grpc.Send

Have you considered using a gofailpoint to simulate watch stream being clogged?

I managed to reproduce the issue using a following test:

Thanks! I have updated the integration test based on the one you provided!

Added unit test, integration test with failpoint and e2e test together and let you guys pick which one is the best.

The e2e test is still a little over complicated to me. It will definitely be painful for other contributors to update/maintain such e2e test in future. I think it should be part of the robustness test, which already supports generating traffic, watching, compaction and verifying results (including watchResponses), why do you have to write similar complex test in e2e test suite? But I won't insist on that.

The e2e test is still a little over complicated to me. It will definitely be painful for other contributors to update/maintain such e2e test in future. I think it should be part of the robustness test, which already supports generating traffic, watching, compaction and verifying results (including watchResponses), why do you have to write similar complex test in e2e test suite? But I won't insist on that.

Okay. This e2e test is removed from the PR and can be added in robustness or performance test suite later. This test was created originally to simulate kubernetes traffic. With unit test and integration test, it's enough to capture regression.

/retest

@chaochn47 do you have bandwidth to backport this to 3.5 and 3.4? We may need to release a 3.4 and 3.5 patch soon.

@chaochn47 do you have bandwidth to backport this to 3.5 and 3.4? We may need to release a 3.4 and 3.5 patch soon.

@ahrtr @jmhbnz sure. Just provided backports PRs. Could you please take a look?

• [release-3.5] backport fix watch event loss after compaction #17612
• [release-3.4] backport fix watch event loss after compaction #17555 #17610