mvcc: restore unsynced watchers

In case syncWatchersLoop() starts before Restore() is called,
watchers already added by that moment are moved to s.synced by the loop.
However, there is a broken logic that moves watchers from s.synced
to s.uncyned without setting keyWatchers of the watcherGroup.
Eventually syncWatchers() fails to pickup those watchers from s.unsynced
and no events are sent to the watchers, because newWatcherBatch() called
in the function uses wg.watcherSetByKey() internally that requires
a proper keyWatchers value.
This commit is contained in:
Iwasaki Yudai 2018-02-05 17:08:24 -08:00 committed by Gyuho Lee
parent 21a1a28c18
commit eaf7d631ad
2 changed files with 34 additions and 25 deletions

View File

@ -192,7 +192,7 @@ func (s *watchableStore) Restore(b backend.Backend) error {
} }
for wa := range s.synced.watchers { for wa := range s.synced.watchers {
s.unsynced.watchers.add(wa) s.unsynced.add(wa)
} }
s.synced = newWatcherGroup() s.synced = newWatcherGroup()
return nil return nil

View File

@ -297,6 +297,8 @@ func TestWatchFutureRev(t *testing.T) {
} }
func TestWatchRestore(t *testing.T) { func TestWatchRestore(t *testing.T) {
test := func(delay time.Duration) func(t *testing.T) {
return func(t *testing.T) {
b, tmpPath := backend.NewDefaultTmpBackend() b, tmpPath := backend.NewDefaultTmpBackend()
s := newWatchableStore(b, &lease.FakeLessor{}, nil) s := newWatchableStore(b, &lease.FakeLessor{}, nil)
defer cleanup(s, b, tmpPath) defer cleanup(s, b, tmpPath)
@ -312,6 +314,8 @@ func TestWatchRestore(t *testing.T) {
w := newStore.NewWatchStream() w := newStore.NewWatchStream()
w.Watch(testKey, nil, rev-1) w.Watch(testKey, nil, rev-1)
time.Sleep(delay)
newStore.Restore(b) newStore.Restore(b)
select { select {
case resp := <-w.Chan(): case resp := <-w.Chan():
@ -328,6 +332,11 @@ func TestWatchRestore(t *testing.T) {
t.Fatal("failed to receive event in 1 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 // TestWatchBatchUnsynced tests batching on unsynced watchers
func TestWatchBatchUnsynced(t *testing.T) { func TestWatchBatchUnsynced(t *testing.T) {