Function TestV3NoEventsLostOnCompact

tests/integration/v3_watch_test.go:1538–1587

TestV3NoEventsLostOnCompact verifies that slow watchers exit with compacted watch response if its next revision of events are compacted and no lost events sent to client.

(t *testing.T)

// TestV3NoEventsLostOnCompact verifies that slow watchers exit with compacted watch response
// if its next revision of events are compacted and no lost events sent to client.
func TestV3NoEventsLostOnCompact(t *testing.T) {
	if integration.ThroughProxy {
		t.Skip("grpc proxy currently does not support requesting progress notifications")
	}
	integration.BeforeTest(t)
	if len(gofail.List()) == 0 {
		t.Skip("please run 'make gofail-enable' before running the test")
	}
	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()

	// sendLoop throughput is rate-limited to 1 event per second
	require.NoError(t, gofail.Enable("beforeSendWatchResponse", `sleep("1s")`))
	wch := client.Watch(ctx, "foo")

	var rev int64
	writeCount := mvcc.ChanBufLen() * 11 / 10
	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("beforeSendWatchResponse"))

	eventCount := 0
	compacted := false
	for resp := range wch {
		err = resp.Err()
		if err != nil {
			if !errors.Is(err, rpctypes.ErrCompacted) {
				t.Fatalf("want watch response err %v but got %v", rpctypes.ErrCompacted, err)
			}
			compacted = true
			break
		}
		eventCount += len(resp.Events)
		if eventCount == writeCount {
			break
		}
	}
	assert.Truef(t, compacted, "Expected stream to get compacted, instead we got %d events out of %d events", eventCount, writeCount)
}