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 {