From e29e855bdf7a06725de784253ed4f98a1d9ecac1 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 12:04:10 +0200 Subject: [PATCH] Name the pulls buckets of the seat's claimants on the check, so the build agent reads them (issue 1212) --- cmd/mesh-controller/check_pulls_test.go | 26 ++++++ cmd/mesh-controller/checks.go | 13 ++- internal/link/build.go | 12 ++- internal/link/builds_nats.go | 2 +- internal/link/checks_nats.go | 90 +++++++++++++++---- internal/link/pull_buckets_test.go | 113 ++++++++++++++++++++++++ 6 files changed, 233 insertions(+), 23 deletions(-) create mode 100644 cmd/mesh-controller/check_pulls_test.go create mode 100644 internal/link/pull_buckets_test.go diff --git a/cmd/mesh-controller/check_pulls_test.go b/cmd/mesh-controller/check_pulls_test.go new file mode 100644 index 00000000..edf479df --- /dev/null +++ b/cmd/mesh-controller/check_pulls_test.go @@ -0,0 +1,26 @@ +package main + +import ( + "reflect" + "testing" + + "github.com/novox/mesh-controller/internal/inventory" +) + +// 1212/AC1 and AC2: the check request names the pulls bucket of each module claiming the delivery seat, in the order +// of the claimants' names. +func TestACheckNamesThePullsBucketsOfTheSeatsClaimants(t *testing.T) { + for name, c := range map[string]struct { + entries []inventory.Entry + want []string + }{ + "only the gate": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate"), namedOnly("mesh-delivery")}, []string{"mesh-delivery-gate_pulls"}}, + "only today's": {[]inventory.Entry{claimingTheSeat("mesh-delivery"), namedOnly("gitea")}, []string{"mesh-delivery_pulls"}}, + "both, name order": {[]inventory.Entry{claimingTheSeat("mesh-delivery-gate"), claimingTheSeat("mesh-delivery")}, []string{"mesh-delivery_pulls", "mesh-delivery-gate_pulls"}}, + "none": {[]inventory.Entry{namedOnly("gitea")}, nil}, + } { + if got := pullsBucketsOf(c.entries); !reflect.DeepEqual(got, c.want) { + t.Errorf("%s: %v, want %v", name, got, c.want) + } + } +} diff --git a/cmd/mesh-controller/checks.go b/cmd/mesh-controller/checks.go index b564f0c7..1af74676 100644 --- a/cmd/mesh-controller/checks.go +++ b/cmd/mesh-controller/checks.go @@ -12,6 +12,7 @@ import ( "strings" "time" + "github.com/novox/mesh-controller/internal/broker" "github.com/novox/mesh-controller/internal/catalogue" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" @@ -305,7 +306,7 @@ func checkRequestFor(ctx context.Context, open *stores, p link.PullUpdated, scop Held: heldBy(ctx), Seats: seatBases(ctx), Source: sourceOnSeat(source), - Check: &link.CheckRequest{Owner: p.Owner, Repo: p.Repo, Number: p.Number, Base: p.Base, Trunk: p.BaseSHA, Ready: p.Ready, + Check: &link.CheckRequest{Pulls: pullsBucketsOf(entries), Owner: p.Owner, Repo: p.Repo, Number: p.Number, Base: p.Base, Trunk: p.BaseSHA, Ready: p.Ready, Paths: p.Paths, Beside: beside, Modules: scope.Modules, Dependents: scope.Dependents, New: scope.New, Manifests: scope.Manifests, Judge: scope.Judge, Builds: builds, Named: namedForTheCheck(ctx, p, builds), Running: runningBy(ctx)}, @@ -661,3 +662,13 @@ func sayChecked(ctx context.Context, c link.Checked) { fmt.Printf("%s: the verdict could not be said, so the pull request is not told it: %v\n", c.ID, err) } } + +// pullsBucketsOf are the buckets of the pull requests' keys a check names for the build agent: one per module +// claiming the delivery seat, in the order of their names (novox/hq ADR 0437 §3, issue 1212). +func pullsBucketsOf(entries []inventory.Entry) []string { + var out []string + for _, m := range claimantsOf(entries, catalogue.DeliverySeat) { + out = append(out, broker.BucketName(m, "pulls")) + } + return out +} diff --git a/internal/link/build.go b/internal/link/build.go index d16601eb..e824d5ab 100644 --- a/internal/link/build.go +++ b/internal/link/build.go @@ -142,10 +142,14 @@ type PullRef struct { // CheckRequest is what a merge check needs beyond the repository and its head. type CheckRequest struct { - Owner string `json:"owner"` - Repo string `json:"repo"` - Number int `json:"number,omitempty"` - Base string `json:"base,omitempty"` + // Pulls are the buckets of the pull requests' keys, one per module claiming the delivery seat, in the order of + // their names (novox/hq ADR 0437 §3): the build agent, which holds no catalogue, reads and watches the key in + // them. Empty from a controller from before it, and then mesh-delivery's. + Pulls []string `json:"pulls,omitempty"` + Owner string `json:"owner"` + Repo string `json:"repo"` + Number int `json:"number,omitempty"` + Base string `json:"base,omitempty"` // Trunk is the base branch's head the check is asked against, as the forge announced it with the head (novox/hq ADR // 0406 §4): with the head, what the pull request's key says the current check is for, and what a waiting request // is dropped by and a running check stopped by when the key moves. diff --git a/internal/link/builds_nats.go b/internal/link/builds_nats.go index b1985f0d..f95a2295 100644 --- a/internal/link/builds_nats.go +++ b/internal/link/builds_nats.go @@ -439,7 +439,7 @@ func (m *natsMachine) judgedAlready(ctx context.Context, r BuildRequest) string return "" } if c.Group == "" { - key, found, err := ReadPullKey(ctx, m.js.Conn(), c.Owner, c.Repo, c.Number) + key, found, err := ReadPullKey(ctx, m.js.Conn(), PullBucketsOf(c), c.Owner, c.Repo, c.Number) if err != nil { fmt.Fprintf(os.Stderr, "could not read the key of %s/%s#%d, so its check runs: %v\n", c.Owner, c.Repo, c.Number, err) } else if found && key.Judged(r.Ref, c.Trunk) { diff --git a/internal/link/checks_nats.go b/internal/link/checks_nats.go index 924af561..7b3064df 100644 --- a/internal/link/checks_nats.go +++ b/internal/link/checks_nats.go @@ -91,28 +91,79 @@ func (k PullKey) Judged(head, trunk string) bool { return judgedBy && sameCommit(k.Head, head) && sameCommit(k.Trunk, trunk) && trunk != "" } -// ReadPullKey reads a pull request's key from mesh-delivery's bucket: false when it has none. -func ReadPullKey(ctx context.Context, conn *nats.Conn, owner, repo string, number int) (PullKey, bool, error) { +// PullBucketsOf are the buckets a check's pull request's key is looked for in: the ones its request names, and +// mesh-delivery's when it names none (a controller from before the request named them). +func PullBucketsOf(c *CheckRequest) []string { + if c == nil || len(c.Pulls) == 0 { + return []string{broker.PullsBucket} + } + return c.Pulls +} + +// ReadPullKey reads a pull request's key from the named buckets in turn: the first that holds it answers, a bucket that +// does not exist is passed over, and false when none holds it. +func ReadPullKey(ctx context.Context, conn *nats.Conn, buckets []string, owner, repo string, number int) (PullKey, bool, error) { api, err := jetstream.New(conn) if err != nil { return PullKey{}, false, err } - kv, err := api.KeyValue(ctx, broker.PullsBucket) - if err != nil { - return PullKey{}, false, fmt.Errorf("the pull requests' keys: %w", err) + return readPullKeyIn(ctx, buckets, func(ctx context.Context, bucket string) (func(context.Context, string) ([]byte, error), error) { + kv, err := api.KeyValue(ctx, bucket) + if err != nil { + return nil, err + } + return func(ctx context.Context, key string) ([]byte, error) { + e, err := kv.Get(ctx, key) + if err != nil { + return nil, err + } + return e.Value(), nil + }, nil + }, owner, repo, number) +} + +// readPullKeyIn is ReadPullKey over buckets opened by open. +func readPullKeyIn(ctx context.Context, buckets []string, + open func(ctx context.Context, bucket string) (func(context.Context, string) ([]byte, error), error), + owner, repo string, number int) (PullKey, bool, error) { + for _, bucket := range buckets { + get, err := open(ctx, bucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + continue + } + if err != nil { + return PullKey{}, false, fmt.Errorf("the pull requests' keys in %s: %w", bucket, err) + } + value, err := get(ctx, PullKeyOf(owner, repo, number)) + if errors.Is(err, jetstream.ErrKeyNotFound) { + continue + } + if err != nil { + return PullKey{}, false, err + } + var key PullKey + if err := json.Unmarshal(value, &key); err != nil { + return PullKey{}, false, fmt.Errorf("the key of %s/%s#%d is not a pull request's: %w", owner, repo, number, err) + } + return key, true, nil } - e, err := kv.Get(ctx, PullKeyOf(owner, repo, number)) - if errors.Is(err, jetstream.ErrKeyNotFound) { - return PullKey{}, false, nil + return PullKey{}, false, nil +} + +// firstBucketThatExists is the first of the buckets that exists, as exists says; an error other than a missing bucket +// stands. +func firstBucketThatExists(ctx context.Context, buckets []string, exists func(ctx context.Context, bucket string) error) (string, error) { + for _, bucket := range buckets { + err := exists(ctx, bucket) + if errors.Is(err, jetstream.ErrBucketNotFound) { + continue + } + if err != nil { + return "", err + } + return bucket, nil } - if err != nil { - return PullKey{}, false, err - } - var key PullKey - if err := json.Unmarshal(e.Value(), &key); err != nil { - return PullKey{}, false, fmt.Errorf("the key of %s/%s#%d is not a pull request's: %w", owner, repo, number, err) - } - return key, true, nil + return "", fmt.Errorf("none of the pull requests' buckets %v exists", buckets) } // WatchPull calls moved once when the pull request's key moves on from what this check was asked for — its head or its @@ -129,7 +180,12 @@ func WatchPull(ctx context.Context, conn *nats.Conn, r BuildRequest, moved func( if err != nil { return nil, err } - kv, err := api.KeyValue(ctx, broker.PullsBucket) + 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) } diff --git a/internal/link/pull_buckets_test.go b/internal/link/pull_buckets_test.go new file mode 100644 index 00000000..6329b7ae --- /dev/null +++ b/internal/link/pull_buckets_test.go @@ -0,0 +1,113 @@ +package link + +import ( + "context" + "encoding/json" + "errors" + "reflect" + "strings" + "testing" + + "github.com/nats-io/nats.go/jetstream" + + "github.com/novox/mesh-controller/internal/broker" +) + +// The build agent reads and watches the pull requests' keys in the buckets the check request names (novox/hq ADR 0437 +// §3, issue 1212): the controller knows the seat's claimants, the build agent holds no catalogue. + +// 1212/AC4: a request naming no bucket reads mesh-delivery's, as before; the field is optional on the wire. +func TestARequestNamingNoPullsBucketUsesMeshDeliverys(t *testing.T) { + if got := PullBucketsOf(&CheckRequest{}); !reflect.DeepEqual(got, []string{broker.PullsBucket}) { + t.Errorf("buckets %v", got) + } + if got := PullBucketsOf(nil); !reflect.DeepEqual(got, []string{broker.PullsBucket}) { + t.Errorf("buckets of no check %v", got) + } + named := &CheckRequest{Pulls: []string{"mesh-delivery-gate_pulls"}} + if got := PullBucketsOf(named); !reflect.DeepEqual(got, named.Pulls) { + t.Errorf("buckets %v", got) + } + raw := mustJSON(t, CheckRequest{Owner: "o"}) + if contains(string(raw), "pulls") { + t.Errorf("an empty field is written: %s", raw) + } +} + +// opened is a set of buckets by name, each holding keys; a bucket not in it does not exist. +func opened(buckets map[string]map[string]string, opens *[]string) func(context.Context, string) (func(context.Context, string) ([]byte, error), error) { + return func(_ context.Context, bucket string) (func(context.Context, string) ([]byte, error), error) { + *opens = append(*opens, bucket) + keys, ok := buckets[bucket] + if !ok { + return nil, jetstream.ErrBucketNotFound + } + return func(_ context.Context, key string) ([]byte, error) { + v, ok := keys[key] + if !ok { + return nil, jetstream.ErrKeyNotFound + } + return []byte(v), nil + }, nil + } +} + +// 1212/AC3: the buckets are read in turn, the first holding the key answers, a missing bucket is passed over, and +// none holding it is no key. +func TestThePullKeyIsReadFromTheNamedBucketsInTurn(t *testing.T) { + key := PullKeyOf("novox", "a", 7) + both := map[string]map[string]string{ + "mesh-delivery_pulls": {key: `{"head":"old"}`}, + "mesh-delivery-gate_pulls": {key: `{"head":"new"}`}, + } + var opens []string + got, found, err := readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(both, &opens), "novox", "a", 7) + if err != nil || !found || got.Head != "new" || len(opens) != 1 { + t.Errorf("first holding answers: %+v %v %v, opened %v", got, found, err, opens) + } + opens = nil + only := map[string]map[string]string{"mesh-delivery_pulls": {key: `{"head":"old"}`}, "mesh-delivery-gate_pulls": {}} + got, found, err = readPullKeyIn(t.Context(), []string{"mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7) + if err != nil || !found || got.Head != "old" || len(opens) != 2 { + t.Errorf("a bucket without the key is passed over: %+v %v %v, opened %v", got, found, err, opens) + } + opens = nil + got, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery_pulls"}, opened(only, &opens), "novox", "a", 7) + if err != nil || !found || got.Head != "old" { + t.Errorf("a missing bucket is passed over: %+v %v %v", got, found, err) + } + _, found, err = readPullKeyIn(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls"}, opened(only, &opens), "novox", "a", 8) + if err != nil || found { + t.Errorf("none holding it: found %v, %v", found, err) + } + boom := errors.New("bus away") + _, _, err = readPullKeyIn(t.Context(), []string{"x"}, func(context.Context, string) (func(context.Context, string) ([]byte, error), error) { + return nil, boom + }, "novox", "a", 7) + if !errors.Is(err, boom) { + t.Errorf("a real failure is an error: %v", err) + } + // The watch takes the first named bucket that exists. + opens = nil + bucket, err := firstBucketThatExists(t.Context(), []string{"gone_pulls", "mesh-delivery-gate_pulls", "mesh-delivery_pulls"}, func(_ context.Context, b string) error { + opens = append(opens, b) + if b == "gone_pulls" { + return jetstream.ErrBucketNotFound + } + return nil + }) + if err != nil || bucket != "mesh-delivery-gate_pulls" || len(opens) != 2 { + t.Errorf("watch bucket %q %v, tried %v", bucket, err, opens) + } +} + +func mustJSON(t *testing.T, v any) string { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return string(raw) +} + +func contains(s, sub string) bool { return strings.Contains(s, sub) }