diff --git a/internal/mvcc/watchable_store.go b/internal/mvcc/watchable_store.go index 34d4d3265..054be1034 100644 --- a/internal/mvcc/watchable_store.go +++ b/internal/mvcc/watchable_store.go @@ -192,7 +192,7 @@ func (s *watchableStore) Restore(b backend.Backend) error { } for wa := range s.synced.watchers { - s.unsynced.watchers.add(wa) + s.unsynced.add(wa) } s.synced = newWatcherGroup() return nil diff --git a/internal/mvcc/watchable_store_test.go b/internal/mvcc/watchable_store_test.go index aa0e29954..4f21aee45 100644 --- a/internal/mvcc/watchable_store_test.go +++ b/internal/mvcc/watchable_store_test.go @@ -297,36 +297,45 @@ func TestWatchFutureRev(t *testing.T) { } func TestWatchRestore(t *testing.T) { - b, tmpPath := backend.NewDefaultTmpBackend() - s := newWatchableStore(b, &lease.FakeLessor{}, nil) - defer cleanup(s, b, tmpPath) + test := func(delay time.Duration) func(t *testing.T) { + return func(t *testing.T) { + b, tmpPath := backend.NewDefaultTmpBackend() + s := newWatchableStore(b, &lease.FakeLessor{}, nil) + defer cleanup(s, b, tmpPath) - testKey := []byte("foo") - testValue := []byte("bar") - rev := s.Put(testKey, testValue, lease.NoLease) + testKey := []byte("foo") + testValue := []byte("bar") + rev := s.Put(testKey, testValue, lease.NoLease) - newBackend, newPath := backend.NewDefaultTmpBackend() - newStore := newWatchableStore(newBackend, &lease.FakeLessor{}, nil) - defer cleanup(newStore, newBackend, newPath) + newBackend, newPath := backend.NewDefaultTmpBackend() + newStore := newWatchableStore(newBackend, &lease.FakeLessor{}, nil) + defer cleanup(newStore, newBackend, newPath) - w := newStore.NewWatchStream() - w.Watch(0, testKey, nil, rev-1) + w := newStore.NewWatchStream() + w.Watch(0, testKey, nil, rev-1) - newStore.Restore(b) - select { - case resp := <-w.Chan(): - if resp.Revision != rev { - t.Fatalf("rev = %d, want %d", resp.Revision, rev) + time.Sleep(delay) + + newStore.Restore(b) + select { + case resp := <-w.Chan(): + if resp.Revision != rev { + t.Fatalf("rev = %d, want %d", resp.Revision, rev) + } + if len(resp.Events) != 1 { + t.Fatalf("failed to get events from the response") + } + if resp.Events[0].Kv.ModRevision != rev { + t.Fatalf("kv.rev = %d, want %d", resp.Events[0].Kv.ModRevision, rev) + } + case <-time.After(time.Second): + t.Fatal("failed to receive event in 1 second.") + } } - if len(resp.Events) != 1 { - t.Fatalf("failed to get events from the response") - } - if resp.Events[0].Kv.ModRevision != rev { - t.Fatalf("kv.rev = %d, want %d", resp.Events[0].Kv.ModRevision, rev) - } - case <-time.After(time.Second): - t.Fatal("failed to receive event in 1 second.") } + + t.Run("Normal", test(0)) + t.Run("RunSyncWatchLoopBeforeRestore", test(time.Millisecond*120)) // longer than default waitDuration } // TestWatchBatchUnsynced tests batching on unsynced watchers