Repository: etcd-io/etcd
File: server/storage/mvcc/watchable_store_test.go (Go)

Code before:

	maxRev := 10
	compactRev := int64(5)
	for i := 0; i < maxRev; i++ {
		s.Put(testKey, testValue, lease.NoLease)
	}
	_, err := s.Compact(traceutil.TODO(), compactRev)
	require.NoErrorf(t, err, "failed to compact kv (%v)", err)

	w := s.NewWatchStream()
	defer w.Close()

	watchers := map[WatchID]int64{
		0: 1,
		1: 1, // create unsyncd watchers with startRev < compactRev
		2: 6, // create unsyncd watchers with compactRev < startRev < currentRev
	}
	for id, startRev := range watchers {
		_, err := w.Watch(id, testKey, nil, startRev)
		require.NoError(t, err)
	}
	// fill up w.Chan() with 1 buf via 2 compacted watch response
	sImpl, ok := s.(*watchableStore)
	if !ok {
		t.Fatal(errors.New("TestWatchNoEventLossOnCompact: needs a WatchableKV implementation"))
	}
	sImpl.syncWatchers([]mvccpb.Event{})

	for len(watchers) > 0 {
		resp := <-w.Chan()
		if resp.CompactRevision != 0 {
			require.Equal(t, resp.CompactRevision, compactRev)
			require.Contains(t, watchers, resp.WatchID)
			delete(watchers, resp.WatchID)
			continue
		}
		nextRev := watchers[resp.WatchID]
		for _, ev := range resp.Events {
			require.Equalf(t, nextRev, ev.Kv.ModRevision, "got event revision %d but want %d for watcher with watch ID %d", ev.Kv.ModRevision, nextRev, resp.WatchID)
			nextRev++
		}
		if nextRev == sImpl.rev()+1 {
			delete(watchers, resp.WatchID)
		}
	}
}

func TestWatchFutureRev(t *testing.T) {
	b, _ := betesting.NewDefaultTmpBackend(t)
	s := New(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
	defer cleanup(s, b)

Review comment (suggestion):
```suggestion
	require.True(t, ok, "TestWatchNoEventLossOnCompact: needs a WatchableKV implementation")
```

Code after:
	maxRev := 10
	compactRev := int64(5)
	for i := 0; i < maxRev; i++ {
		s.Put(testKey, testValue, lease.NoLease)
	}
	_, err := s.Compact(traceutil.TODO(), compactRev)
	require.NoErrorf(t, err, "failed to compact kv (%v)", err)

	w := s.NewWatchStream()
	defer w.Close()

	watchers := map[WatchID]int64{
		0: 1,
		1: 1, // create unsyncd watchers with startRev < compactRev
		2: 6, // create unsyncd watchers with compactRev < startRev < currentRev
	}
	for id, startRev := range watchers {
		_, err := w.Watch(id, testKey, nil, startRev)
		require.NoError(t, err)
	}
	// fill up w.Chan() with 1 buf via 2 compacted watch response
	sImpl, ok := s.(*watchableStore)
	require.Truef(t, ok, "TestWatchNoEventLossOnCompact: needs a WatchableKV implementation")
	sImpl.syncWatchers([]mvccpb.Event{})

	for len(watchers) > 0 {
		resp := <-w.Chan()
		if resp.CompactRevision != 0 {
			require.Equal(t, resp.CompactRevision, compactRev)
			require.Contains(t, watchers, resp.WatchID)
			delete(watchers, resp.WatchID)
			continue
		}
		nextRev := watchers[resp.WatchID]
		for _, ev := range resp.Events {
			require.Equalf(t, nextRev, ev.Kv.ModRevision, "got event revision %d but want %d for watcher with watch ID %d", ev.Kv.ModRevision, nextRev, resp.WatchID)
			nextRev++
		}
		if nextRev == sImpl.rev()+1 {
			delete(watchers, resp.WatchID)
		}
	}
}

func TestWatchFutureRev(t *testing.T) {
	b, _ := betesting.NewDefaultTmpBackend(t)
	s := New(zaptest.NewLogger(t), b, &lease.FakeLessor{}, StoreConfig{})
	defer cleanup(s, b)

	testKey := []byte("foo")
	testValue := []byte("bar")