Mirroristas/etcd
2
0
Fork 0
mirror of https://github.com/etcd-io/etcd.git synced 2024-09-27 06:25:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #15515 from serathius/fix-issue15271-3.5

Browse Source 
[v3.5] Fix issue15271
This commit is contained in:
Marek Siarkowicz 2023-03-20 13:08:31 +01:00 committed by GitHub
commit 15b3756abd
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 27 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
server/mvcc/watchable_store.go
View File

@ -447,7 +447,6 @@ func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) {
pendingEventsGauge.Add(float64(len(eb.evs)))
} else {
// move slow watcher to victims
w.minRev = rev + 1
if victim == nil {
victim = make(watcherBatch)
}
@ -456,6 +455,10 @@ func (s *watchableStore) notify(rev int64, evs []mvccpb.Event) {
s.synced.delete(w)
slowWatcherGauge.Inc()
}
// always update minRev
// in case 'send' returns true and watcher stays synced, this is needed for Restore when all watchers become unsynced
// in case 'send' returns false, this is needed for syncWatchers
w.minRev = rev + 1
}
s.addVictim(victim)
}

45
server/mvcc/watchable_store_test.go
View File

@ -307,32 +307,22 @@ func TestWatchRestore(t *testing.T) {
testKey := []byte("foo")
testValue := []byte("bar")
rev := s.Put(testKey, testValue, lease.NoLease)
newBackend, newPath := betesting.NewDefaultTmpBackend(t)
newStore := newWatchableStore(zap.NewExample(), newBackend, &lease.FakeLessor{}, StoreConfig{})
defer cleanup(newStore, newBackend, newPath)
w := newStore.NewWatchStream()
w.Watch(0, testKey, nil, rev-1)
w := s.NewWatchStream()
defer w.Close()
w.Watch(0, testKey, nil, 1)
time.Sleep(delay)
wantRev := s.Put(testKey, testValue, lease.NoLease)
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.")
s.Restore(b)
events := readEventsForSecond(w.Chan())
if len(events) != 1 {
t.Errorf("Expected only one event, got %d", len(events))
}
if events[0].Kv.ModRevision != wantRev {
t.Errorf("Expected revision to match, got %d, want %d", events[0].Kv.ModRevision, wantRev)
}
}
}
@ -340,6 +330,17 @@ func TestWatchRestore(t *testing.T) {
t.Run("RunSyncWatchLoopBeforeRestore", test(time.Millisecond*120)) // longer than default waitDuration
}
func readEventsForSecond(ws <-chan WatchResponse) (events []mvccpb.Event) {
for {
select {
case resp := <-ws:
events = append(events, resp.Events...)
case <-time.After(time.Second):
return events
}
}
}
// TestWatchRestoreSyncedWatcher tests such a case that:
// 1. watcher is created with a future revision "math.MaxInt64 - 2"
// 2. watcher with a future revision is added to "synced" watcher group