diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index 7b3064df..176c70ec 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -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 } diff --git a/internal/link/pull_buckets_test.go b/internal/link/pull_buckets_test.go index 6329b7ae..2e27f187 100644 --- a/internal/link/pull_buckets_test.go +++ b/internal/link/pull_buckets_test.go @@ -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") + } +} diff --git a/modules/build-agent/manifest_test.go b/modules/build-agent/manifest_test.go index 59e041d9..a0716251 100644 --- a/modules/build-agent/manifest_test.go +++ b/modules/build-agent/manifest_test.go @@ -13,6 +13,7 @@ import ( ) type manifest struct { + Reads []string `json:"reads"` Requires []json.RawMessage `json:"requires"` Binds map[string]string `json:"binds"` Secrets map[string]string `json:"secrets"` @@ -172,3 +173,17 @@ func TestTheJudgeCacheIsADirectoryOfItsOwnMountedAtTheSamePath(t *testing.T) { t.Fatalf("the judge cache: directory %v, data %v, env %v, mounted at the same path %v", dir, data, env, mounted) } } + +// A running check's watch reads the pull requests' keys in the gate's bucket and in mesh-delivery's, so it +// stays sighted whichever of them the keys are written to (novox/hq ADR 0450 §1, issue 1325 AC1). +func TestTheAgentReadsTheGatesPullKeysAndTodaysBoth(t *testing.T) { + got := map[string]bool{} + for _, r := range read(t).Reads { + got[r] = true + } + for _, want := range []string{"mesh-delivery-gate.pulls", "mesh-delivery.pulls"} { + if !got[want] { + t.Errorf("the build agent does not read %s: %v", want, read(t).Reads) + } + } +} diff --git a/modules/build-agent/module.json b/modules/build-agent/module.json index 4d74764d..5a550ef6 100644 --- a/modules/build-agent/module.json +++ b/modules/build-agent/module.json @@ -13,6 +13,7 @@ } ], "reads": [ + "mesh-delivery-gate.pulls", "mesh-delivery.pulls" ], "requires": [