Repository: etcd-io/etcd
File: tests/integration/clientv3/watch/v3_watch_test.go (Go)

Code before:
		select {
		case <-ticker.C:
			err := client.RequestProgress(ctx)
			require.NoError(t, err)
		case resp := <-wch:
			if resp.Err() != nil {
				t.Fatal(fmt.Errorf("watch error: %w", resp.Err()))
			}
			if resp.IsProgressNotify() {
				gotProgressNotification = true
			}
		}
		if gotProgressNotification {
			break
		}
	}
	require.Truef(t, gotProgressNotification, "Expected to get progress notification")
}

// 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)
	integration.GofailSkipCheck(t)
	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)

	client := clus.RandClient()
	ctx, cancel := context.WithTimeout(t.Context(), 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"))


Review comment (bug):
```suggestion
	integration.BeforeTest(t, WithFailpoint("beforeSendWatchResponse"))
```

Code after:
		select {
		case <-ticker.C:
			err := client.RequestProgress(ctx)
			require.NoError(t, err)
		case resp := <-wch:
			if resp.Err() != nil {
				t.Fatal(fmt.Errorf("watch error: %w", resp.Err()))
			}
			if resp.IsProgressNotify() {
				gotProgressNotification = true
			}
		}
		if gotProgressNotification {
			break
		}
	}
	require.Truef(t, gotProgressNotification, "Expected to get progress notification")
}

// 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)
	integration.SkipIfNoGoFail(t)
	clus := integration.NewCluster(t, &integration.ClusterConfig{Size: 1})
	defer clus.Terminate(t)

	client := clus.RandClient()
	ctx, cancel := context.WithTimeout(t.Context(), 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"))