diff --git a/internal/link/pull_buckets_test.go b/internal/link/pull_buckets_test.go index 2e27f187..7a98a939 100644 --- a/internal/link/pull_buckets_test.go +++ b/internal/link/pull_buckets_test.go @@ -6,6 +6,7 @@ import ( "errors" "reflect" "strings" + "sync" "testing" "time" @@ -116,11 +117,30 @@ func contains(s, sub string) bool { return strings.Contains(s, sub) } // fakeWatch is one bucket's watch of the key, fed by the test. type fakeWatch struct { updates chan jetstream.KeyValueEntry - stopped bool + done chan struct{} // closed when the watch is stopped + once sync.Once +} + +func newFakeWatch() *fakeWatch { + return &fakeWatch{updates: make(chan jetstream.KeyValueEntry, 1), done: make(chan struct{})} +} + +// handled returns once the watch has taken the entry fed before it and finished with it. The channel holds one entry, +// so the second filler (an entry the watch passes over) can only be sent after the first was taken, which the watch +// does only after it handled the entry before it. It needs a watch that is still listening: after a move it is not. +func (f *fakeWatch) handled(t *testing.T) { + t.Helper() + for range 2 { + select { + case f.updates <- fakeEntry{value: "not a key"}: + case <-time.After(2 * time.Second): + t.Fatal("the watch took no entry") + } + } } func (f *fakeWatch) Updates() <-chan jetstream.KeyValueEntry { return f.updates } -func (f *fakeWatch) Stop() error { f.stopped = true; return nil } +func (f *fakeWatch) Stop() error { f.once.Do(func() { close(f.done) }); return nil } type fakeEntry struct { jetstream.KeyValueEntry @@ -138,7 +158,7 @@ func TestARunningChecksWatchFindsItsKeyInTheGatesBucketOrTodays(t *testing.T) { run := func(exist ...string) (map[string]*fakeWatch, chan PullKey, func(), error) { watches := map[string]*fakeWatch{} for _, b := range exist { - watches[b] = &fakeWatch{updates: make(chan jetstream.KeyValueEntry, 1)} + watches[b] = newFakeWatch() } moved := make(chan PullKey, 2) stop, err := watchPullIn(t.Context(), req.Check.Pulls, func(_ context.Context, bucket string) (pullWatcher, error) { @@ -182,14 +202,28 @@ func TestARunningChecksWatchFindsItsKeyInTheGatesBucketOrTodays(t *testing.T) { t.Fatal(err) } watches["mesh-delivery-gate_pulls"].updates <- fakeEntry{value: `{"head":"old","trunk":"t","verdict":"pass"}`} - if _, ok := wait(moved); ok { + watches["mesh-delivery-gate_pulls"].handled(t) + select { + case <-moved: t.Error("a verdict alone moved the check") + default: } + // Both buckets hear the move; the second watch is done (stopped) once it has handled its entry, and by then a + // second telling would be in the channel. watches["mesh-delivery-gate_pulls"].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`} watches["mesh-delivery_pulls"].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`} wait(moved) - if _, ok := wait(moved); ok { + for b, w := range watches { + select { + case <-w.done: + case <-time.After(2 * time.Second): + t.Errorf("the watch of %s did not end after the move", b) + } + } + select { + case <-moved: t.Error("a move was told twice") + default: } stop() if _, _, _, err := run(); err == nil {