Let the build agent read the gate's pull keys and watch both buckets (issue 1325)

From R3e the gate writes mesh-delivery-gate_pulls and today's bucket stops (ADR 0450 §1). The build agent's manifest reads both, and a running check's watch listens in every bucket that exists, so a move is heard whichever holds the key.
2026-10-19 21:42:37 +00:00
parent 23a209f3d0
commit 4083755ca1
4 changed files with 164 additions and 34 deletions
@@ -8,6 +8,7 @@ import (
 	"os"
 	"strconv"
 	"strings"
+	"sync"
 	"time"

 	"github.com/nats-io/nats.go"
@@ -180,45 +181,73 @@ func WatchPull(ctx context.Context, conn *nats.Conn, r BuildRequest, moved func(
 	if err != nil {
 		return nil, err
 	}
-	var kv jetstream.KeyValue
-	_, err = firstBucketThatExists(ctx, PullBucketsOf(c), func(ctx context.Context, bucket string) error {
-		var err error
-		kv, err = api.KeyValue(ctx, bucket)
-		return err
-	})
-	if err != nil {
-		return nil, fmt.Errorf("the pull requests' keys: %w", err)
-	}
+	return watchPullIn(ctx, PullBucketsOf(c), func(ctx context.Context, bucket string) (pullWatcher, error) {
+		kv, err := api.KeyValue(ctx, bucket)
+		if err != nil {
+			return nil, err
+		}
+		return kv.Watch(ctx, PullKeyOf(c.Owner, c.Repo, c.Number), jetstream.UpdatesOnly())
+	}, r, moved)
+}
+
+// pullWatcher is what a watch of one bucket's key gives: the updates, and a stop.
+type pullWatcher interface {
+	Updates() <-chan jetstream.KeyValueEntry
+	Stop() error
+}
+
+// watchPullIn is WatchPull over buckets watched by open: the key is watched in every bucket that exists, since
+// the gate writes its own bucket and mesh-delivery's stops only when the gate takes over (novox/hq ADR 0450 §1,
+// issue 1325); a move in any of them stops the check, once. None existing is an error.
+func watchPullIn(ctx context.Context, buckets []string, open func(ctx context.Context, bucket string) (pullWatcher, error),
+	r BuildRequest, moved func(PullKey)) (stop func(), err error) {
 	watching, end := context.WithCancel(ctx)
-	w, err := kv.Watch(watching, PullKeyOf(c.Owner, c.Repo, c.Number), jetstream.UpdatesOnly())
-	if err != nil {
-		end()
-		return nil, err
+	var watchers []pullWatcher
+	for _, bucket := range buckets {
+		w, err := open(watching, bucket)
+		if errors.Is(err, jetstream.ErrBucketNotFound) {
+			continue
+		}
+		if err != nil {
+			end()
+			for _, held := range watchers {
+				_ = held.Stop()
+			}
+			return nil, fmt.Errorf("the pull requests' keys in %s: %w", bucket, err)
+		}
+		watchers = append(watchers, w)
 	}
-	go func() {
-		defer w.Stop()
-		for {
-			select {
-			case <-watching.Done():
-				return
-			case e, ok := <-w.Updates():
-				if !ok {
-					return
-				}
-				if e == nil || e.Operation() != jetstream.KeyValuePut {
-					continue
-				}
-				var key PullKey
-				if json.Unmarshal(e.Value(), &key) != nil {
-					continue
-				}
-				if key.Moved(r.Ref, c.Trunk) {
-					moved(key)
+	if len(watchers) == 0 {
+		end()
+		return nil, fmt.Errorf("the pull requests' keys: none of the buckets %v exists", buckets)
+	}
+	var once sync.Once
+	for _, w := range watchers {
+		go func(w pullWatcher) {
+			defer w.Stop()
+			for {
+				select {
+				case <-watching.Done():
 					return
+				case e, ok := <-w.Updates():
+					if !ok {
+						return
+					}
+					if e == nil || e.Operation() != jetstream.KeyValuePut {
+						continue
+					}
+					var key PullKey
+					if json.Unmarshal(e.Value(), &key) != nil {
+						continue
+					}
+					if key.Moved(r.Ref, r.Check.Trunk) {
+						once.Do(func() { moved(key) })
+						return
+					}
 				}
 			}
-		}
-	}()
+		}(w)
+	}
 	return end, nil
 }

@@ -7,6 +7,7 @@ import (
 	"reflect"
 	"strings"
 	"testing"
+	"time"

 	"github.com/nats-io/nats.go/jetstream"

@@ -111,3 +112,87 @@ func mustJSON(t *testing.T, v any) string {
 }

 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
+}
+
+func (f *fakeWatch) Updates() <-chan jetstream.KeyValueEntry { return f.updates }
+func (f *fakeWatch) Stop() error                             { f.stopped = true; return nil }
+
+type fakeEntry struct {
+	jetstream.KeyValueEntry
+	value string
+}
+
+func (e fakeEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
+func (e fakeEntry) Value() []byte                   { return []byte(e.value) }
+
+// 1325/AC2: a running check's watch hears the key move in the gate's bucket when it is there, in today's when only
+// it is there, and in either when both are; a bucket that does not exist is passed over, and none is an error.
+func TestARunningChecksWatchFindsItsKeyInTheGatesBucketOrTodays(t *testing.T) {
+	req := BuildRequest{Ref: "old", Check: &CheckRequest{Owner: "novox", Repo: "a", Number: 7, Trunk: "t",
+		Pulls: []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}}}
+	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)}
+		}
+		moved := make(chan PullKey, 2)
+		stop, err := watchPullIn(t.Context(), req.Check.Pulls, func(_ context.Context, bucket string) (pullWatcher, error) {
+			if w, ok := watches[bucket]; ok {
+				return w, nil
+			}
+			return nil, jetstream.ErrBucketNotFound
+		}, req, func(k PullKey) { moved <- k })
+		return watches, moved, stop, err
+	}
+	wait := func(moved chan PullKey) (PullKey, bool) {
+		select {
+		case k := <-moved:
+			return k, true
+		case <-time.After(2 * time.Second):
+			return PullKey{}, false
+		}
+	}
+	for name, c := range map[string]struct {
+		exist []string
+		feed  string
+	}{
+		"only the gate's":  {[]string{"mesh-delivery-gate_pulls"}, "mesh-delivery-gate_pulls"},
+		"only today's":     {[]string{"mesh-delivery_pulls"}, "mesh-delivery_pulls"},
+		"both, the gate's": {[]string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, "mesh-delivery-gate_pulls"},
+		"both, today's":    {[]string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, "mesh-delivery_pulls"},
+	} {
+		watches, moved, stop, err := run(c.exist...)
+		if err != nil {
+			t.Fatalf("%s: %v", name, err)
+		}
+		watches[c.feed].updates <- fakeEntry{value: `{"head":"newer","trunk":"t"}`}
+		if k, ok := wait(moved); !ok || k.Head != "newer" {
+			t.Errorf("%s: the move was not heard (%+v, %v)", name, k, ok)
+		}
+		stop()
+	}
+	// A key that only gains its verdict is no move, and a move is told once.
+	watches, moved, stop, err := run("mesh-delivery-gate_pulls", "mesh-delivery_pulls")
+	if err != nil {
+		t.Fatal(err)
+	}
+	watches["mesh-delivery-gate_pulls"].updates <- fakeEntry{value: `{"head":"old","trunk":"t","verdict":"pass"}`}
+	if _, ok := wait(moved); ok {
+		t.Error("a verdict alone moved the check")
+	}
+	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 {
+		t.Error("a move was told twice")
+	}
+	stop()
+	if _, _, _, err := run(); err == nil {
+		t.Error("no bucket existing was not an error")
+	}
+}