From 5c606fd1d48b49b128ccdcd1205f15addbb4394a Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 13 Oct 2026 23:16:55 +0200 Subject: [PATCH] Take only a check build its verdict names, and keep a build another check still names The artifact store takes writes from anything on its network, so a record planted under another merge's key could have been registered: the walk now takes a scratch build only when the build seat's own verdict for one of its pull requests names its key and digests. Deleting by digest also took builds another check shared (review of #311; novox/hq ADR 0350). --- cmd/mesh-builder/main.go | 10 +++ cmd/mesh-controller/check_builds_test.go | 26 ++++++ cmd/mesh-controller/release_plan.go | 30 ++++++- internal/builder/builder.go | 8 ++ internal/builder/reuse.go | 81 ++++++++++++++++-- internal/builder/reuse_test.go | 102 +++++++++++++++++++++-- internal/link/build.go | 16 +++- internal/link/check_builds_test.go | 39 +++++++++ internal/link/queue.go | 58 +++++++++++++ 9 files changed, 354 insertions(+), 16 deletions(-) create mode 100644 internal/link/check_builds_test.go diff --git a/cmd/mesh-builder/main.go b/cmd/mesh-builder/main.go index 4254462b..5f99bd65 100644 --- a/cmd/mesh-builder/main.go +++ b/cmd/mesh-builder/main.go @@ -311,6 +311,16 @@ func answer(ctx context.Context, publisher builder.Publisher, on, workspace stri for _, p := range request.Reuse.Pulls { reuse.Pulls = append(reuse.Pulls, builder.PullRef{Owner: p.Owner, Repo: p.Repo, Number: p.Number}) } + // Only what a verdict names is taken (ADR 0350 §2). + for _, b := range request.Reuse.Builds { + named := builder.NamedBuild{Pull: builder.PullRef{Owner: b.Pull.Owner, Repo: b.Pull.Repo, + Number: b.Pull.Number}, Check: b.Check, Module: b.Built.Module, Key: b.Built.Key, + Digests: map[string]string{}} + for _, m := range b.Built.Made { + named.Digests[m.Name] = builder.DigestOf(m.Reference) + } + reuse.Named = append(reuse.Named, named) + } built, err = builder.BuildReusing(building, builder.Labelled(builder.Command, request.ID), publisher, reuse, request.Repository, request.Path, request.Ref, workspace, request.Held, npmrc, forgeFrom(), say, request.Seats) diff --git a/cmd/mesh-controller/check_builds_test.go b/cmd/mesh-controller/check_builds_test.go index a4dda987..d0f41d6a 100644 --- a/cmd/mesh-controller/check_builds_test.go +++ b/cmd/mesh-controller/check_builds_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "reflect" "strings" "testing" @@ -188,3 +189,28 @@ func TestAReusedCheckBuildIsRegisteredByItsDigestAndGatedAsABuiltOne(t *testing. t.Fatalf("the first machine was sent %v, not the reused build", sent) } } + +// A walk's leave names the builds its own pull requests' verdicts made, and none when they cannot be read: the +// builder takes nothing a verdict does not name (novox/hq ADR 0350 §2). +func TestAWalksLeaveNamesTheBuildsItsPullRequestsVerdictsMade(t *testing.T) { + was := checkedBuildsOf + t.Cleanup(func() { checkedBuildsOf = was }) + var asked []link.PullRef + named := []link.CheckedBuild{{Pull: link.PullRef{Owner: "novox", Repo: "shop", Number: 7}, Check: "build-1", + Built: link.CheckBuilt{Module: "app", Key: "r1:1"}}} + checkedBuildsOf = func(_ context.Context, pulls []link.PullRef) ([]link.CheckedBuild, error) { + asked = pulls + return named, nil + } + pulls := []link.PullRef{{Owner: "novox", Repo: "shop", Number: 7}} + if got := reuseWithVerdicts(t.Context(), &link.ReuseAsk{Pulls: pulls}); !reflect.DeepEqual(got.Builds, named) || + !reflect.DeepEqual(asked, pulls) { + t.Fatalf("the leave names %+v, read for %+v", got.Builds, asked) + } + checkedBuildsOf = func(context.Context, []link.PullRef) ([]link.CheckedBuild, error) { + return nil, errors.New("no bus") + } + if got := reuseWithVerdicts(t.Context(), &link.ReuseAsk{Pulls: pulls}); got.Builds != nil { + t.Fatalf("verdicts that could not be read named %+v", got.Builds) + } +} diff --git a/cmd/mesh-controller/release_plan.go b/cmd/mesh-controller/release_plan.go index e2d30cdf..77060d8b 100644 --- a/cmd/mesh-controller/release_plan.go +++ b/cmd/mesh-controller/release_plan.go @@ -13,6 +13,7 @@ import ( "github.com/nats-io/nats.go" + "github.com/novox/mesh-controller/internal/builder" "github.com/novox/mesh-controller/internal/inventory" "github.com/novox/mesh-controller/internal/link" ) @@ -356,6 +357,33 @@ func reuseAskOf(p *inventory.Plan) *link.ReuseAsk { return r } +// checkedBuildsOf are the builds the build seat's checks of those pull requests said they made for their merges, +// within the week check builds are kept (novox/hq ADR 0350 §2): read from the seat's own outcomes on the bus. A +// variable so a test needs no bus. +var checkedBuildsOf = func(ctx context.Context, pulls []link.PullRef) ([]link.CheckedBuild, error) { + if len(pulls) == 0 { + return nil, nil + } + js, err := aBus() + if err != nil { + return nil, err + } + defer js.Close() + return link.ReadCheckBuilds(ctx, js, buildSeatHeld(ctx), pulls, time.Now().Add(-builder.CheckBuildsKept)) +} + +// reuseWithVerdicts is a walk's leave with the builds its pull requests' verdicts name: the only builds the builder +// may take (ADR 0350 §2). When they cannot be read the leave names none, and the walk builds, saying why. +func reuseWithVerdicts(ctx context.Context, r *link.ReuseAsk) *link.ReuseAsk { + builds, err := checkedBuildsOf(ctx, r.Pulls) + if err != nil { + fmt.Printf(" the checks' builds of %d pull request(s) could not be read, so the walk builds: %v\n", len(r.Pulls), err) + return r + } + r.Builds = builds + return r +} + // pullOf is a merge's pull request: owner and repository, the last two parts of however the repository is spelled // (owner/repo, a path on the git seat, a URL), and its number. False when the merge names none. func pullOf(m inventory.PlanMerge) (link.PullRef, bool) { @@ -395,7 +423,7 @@ func askModule(ctx context.Context, p *inventory.Plan, name string, byName map[s } // **The walk takes the check's build of the same key** (novox/hq ADR 0350 §2): its ask says so, with the pull // requests it answers, whose checks name the part of a differing key. - id, err := askABuild(withReuse(ctx, reuseAskOf(p)), source, e.Source.Path, ref) + id, err := askABuild(withReuse(ctx, reuseWithVerdicts(ctx, reuseAskOf(p))), source, e.Source.Path, ref) if err != nil { state.State = "failed" state.Why = err.Error() diff --git a/internal/builder/builder.go b/internal/builder/builder.go index 0ff70c71..e819ea41 100644 --- a/internal/builder/builder.go +++ b/internal/builder/builder.go @@ -156,6 +156,9 @@ func Build(ctx context.Context, run Runner, publish Publisher, type Reuse struct { Scratch Scratch Pulls []PullRef + // Named are the builds the verdicts of the walk's own pull requests name (novox/hq ADR 0350 §2): a scratch build + // is taken only when one of them names its key and every digest. The record in the store alone is never enough. + Named []NamedBuild // keyFirst is a check's build (§1): its key is taken before anything is built, and a module with none is not // built at all — nothing could ever take it. keyFirst bool @@ -465,6 +468,11 @@ func takeACheckBuild(ctx context.Context, run Runner, publish Publisher, reuse R if !ok { return nil, nil, notKeptWhy(ctx, reuse, manifest.Module, key) } + // **Only a build a verdict names** (ADR 0350 §2): the store takes writes from anything on the network, so a + // record is believed only when a check verdict of the walk's own pull requests names its key and its digests. + if why := unnamed(reuse.Named, manifest.Module, key.Key, found); why != "" { + return nil, nil, why + } taken := &Reused{From: found} for _, made := range found.Made { copied, err := reuse.Scratch.Take(ctx, manifest.Module, made) diff --git a/internal/builder/reuse.go b/internal/builder/reuse.go index ad832e48..04a09ce0 100644 --- a/internal/builder/reuse.go +++ b/internal/builder/reuse.go @@ -442,11 +442,17 @@ func (r Registry) blobSize(ctx context.Context, repository, digest string) (int6 // Forget lets go of every check build of a pull request (§5): what each check built in the scratch namespace, the // record under its key when it is still that check's, and the check's document. func (r Registry) Forget(ctx context.Context, pull PullRef) (int, error) { - checks, err := r.checks(ctx, []PullRef{pull}) + all, err := r.checks(ctx, nil) if err != nil { return 0, err } - return r.letGo(ctx, checks) + mine := map[string]checkIndex{} + for tag, c := range all { + if c.Pull == pull { + mine[tag] = c + } + } + return r.letGo(ctx, mine, all) } // Expire lets go of every check older than the bound (§5). @@ -461,19 +467,36 @@ func (r Registry) Expire(ctx context.Context, older time.Duration, now time.Time old[tag] = c } } - return r.letGo(ctx, old) + return r.letGo(ctx, old, checks) } // Discard lets go of builds no check document names. func (r Registry) Discard(ctx context.Context, builds []CheckBuild) error { - _, err := r.letGo(ctx, map[string]checkIndex{"": {Builds: builds}}) + all, err := r.checks(ctx, nil) + if err != nil { + return err + } + _, err = r.letGo(ctx, map[string]checkIndex{"": {Builds: builds}}, all) return err } // letGo lets go of each check's builds, then its document; a check whose builds could not all be let go of keeps -// its document, so the next sweep asks again. -func (r Registry) letGo(ctx context.Context, checks map[string]checkIndex) (int, error) { +// its document, so the next sweep asks again. **A build another check still names stays**: deletion is by digest, +// and two checks that built one digest in one repository — a check asked again, a dependent two pull requests move — +// share it; it goes with the last check that names it. all are every check kept, the ones going among them. +func (r Registry) letGo(ctx context.Context, checks, all map[string]checkIndex) (int, error) { store := r.store() + stays := map[string]bool{} + for tag, c := range all { + if _, going := checks[tag]; going { + continue + } + for _, b := range c.Builds { + for _, m := range b.Made { + stays[catalogue.Recorded(m.Reference)] = true + } + } + } gone := 0 var failed []string for tag, c := range checks { @@ -484,6 +507,9 @@ func (r Registry) letGo(ctx context.Context, checks map[string]checkIndex) (int, if path, kept := catalogue.InArtifactStore(reference); !kept || !strings.HasPrefix(path, ScratchPrefix) { continue // never anything outside the scratch namespace } + if stays[reference] { + continue // another check still names it + } if err := store.LetGo(ctx, reference); err != nil && !errors.Is(err, artifacts.Gone) { ok = false failed = append(failed, fmt.Sprintf("%s: %v", m.Reference, err)) @@ -537,3 +563,46 @@ func (r Reused) Words() string { } return "reused from " + r.From.Said() + ": " + strings.Join(digests, ", ") } + +// NamedBuild is a build a check's verdict names (ADR 0350 §2): its module, key and each artifact's digest. +type NamedBuild struct { + Pull PullRef + Check string + Module string + Key string + Digests map[string]string +} + +// DigestOf is the digest a reference names: an image's after `@`, an archive's after `/blobs/`. +func DigestOf(reference string) string { + if _, d, ok := strings.Cut(reference, "/blobs/"); ok { + return d + } + if _, d, ok := strings.Cut(reference, "@"); ok { + return d + } + return "" +} + +// unnamed is why a record found under a key is not taken: no verdict of the walk's pull requests names that module, +// key and every one of its digests. Empty when one does. +func unnamed(named []NamedBuild, module, key string, found CheckBuild) string { + for _, n := range named { + if n.Module != module || n.Key != key || n.Check != found.Check || len(n.Digests) != len(found.Made) { + continue + } + same := true + for _, m := range found.Made { + d := m.Digest + if d == "" { + d = DigestOf(m.Reference) + } + same = same && d != "" && n.Digests[m.Name] == d + } + if same { + return "" + } + } + return "a build is kept under this key, and no check verdict of the walk's pull requests names it with its " + + "digests: not taken" +} diff --git a/internal/builder/reuse_test.go b/internal/builder/reuse_test.go index a3e3c1f5..e3e4ec24 100644 --- a/internal/builder/reuse_test.go +++ b/internal/builder/reuse_test.go @@ -124,6 +124,19 @@ func aCheckBuilt(t *testing.T, scratch *aScratch, tree string) CheckBuilt { return built[0] } +// named is the walk's leave naming what a check's verdict says it built, as the controller reads it. +func named(c CheckBuilt) []NamedBuild { + n := NamedBuild{Pull: aPull, Check: "build-1", Module: c.Module, Key: c.Key, Digests: map[string]string{}} + for _, m := range c.Made { + d := m.Digest + if d == "" { + d = DigestOf(m.Reference) + } + n.Digests[m.Name] = d + } + return []NamedBuild{n} +} + func withVersion(t *testing.T, v string) { t.Helper() was := Version @@ -139,7 +152,7 @@ func TestAWalkWhoseKeyIsItsChecksBuildsNothingAndRegistersTheChecksDigest(t *tes checked := aCheckBuilt(t, scratch, "tree-one") r, run, workspace := aModuleAt(t, "tree-one") - got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}}, + got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}, Named: named(checked)}, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) if err != nil { t.Fatal(err) @@ -176,14 +189,47 @@ func TestAWalkWhoseKeyIsItsChecksBuildsNothingAndRegistersTheChecksDigest(t *tes } } +// A record no verdict names is refused and built instead, and the walk says so: the store takes writes from anything +// on its network, so a record planted under another merge's key is never registered. +func TestARecordNoVerdictNamesIsRefusedAndBuilt(t *testing.T) { + withVersion(t, "builder-1") + scratch := newScratch() + checked := aCheckBuilt(t, scratch, "tree-one") + for name, leave := range map[string][]NamedBuild{ + "no verdict": nil, + "another digest": func() []NamedBuild { + n := named(checked) + n[0].Digests["server"] = "sha256:" + strings.Repeat("e", 64) + return n + }(), + "another check": func() []NamedBuild { n := named(checked); n[0].Check = "build-9"; return n }(), + } { + t.Run(name, func(t *testing.T) { + scratch.took = nil + r, run, workspace := aModuleAt(t, "tree-one") + got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}, Named: leave}, + "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) + if err != nil { + t.Fatal(err) + } + if got.Reused != nil || len(scratch.took) > 0 || !dockerBuilt(r) { + t.Fatalf("a record no verdict names was taken: %+v, copied %v", got.Reused, scratch.took) + } + if !strings.Contains(got.Why, "no check verdict of the walk's pull requests names it") { + t.Fatalf("the walk says %q", got.Why) + } + }) + } +} + // A differing key builds, and says which part differs. func TestAWalkWhoseKeyDiffersBuildsAndSaysWhy(t *testing.T) { withVersion(t, "builder-1") scratch := newScratch() - aCheckBuilt(t, scratch, "tree-one") + checked := aCheckBuilt(t, scratch, "tree-one") r, run, workspace := aModuleAt(t, "tree-two") - got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}}, + got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}, Named: named(checked)}, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) if err != nil { t.Fatal(err) @@ -198,7 +244,7 @@ func TestAWalkWhoseKeyDiffersBuildsAndSaysWhy(t *testing.T) { // And a builder of another version builds too, saying so. withVersion(t, "builder-2") r, run, workspace = aModuleAt(t, "tree-one") - got, err = BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}}, + got, err = BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}, Named: named(checked)}, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) if err != nil { t.Fatal(err) @@ -220,11 +266,11 @@ func TestAWalkWhoseKeyDiffersBuildsAndSaysWhy(t *testing.T) { func TestAFailedCopyOfAChecksBuildFallsBackToBuilding(t *testing.T) { withVersion(t, "builder-1") scratch := newScratch() - aCheckBuilt(t, scratch, "tree-one") + checked := aCheckBuilt(t, scratch, "tree-one") scratch.failTake = true r, run, workspace := aModuleAt(t, "tree-one") - got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}}, + got, err := BuildReusing(t.Context(), run, r, Reuse{Scratch: scratch, Pulls: []PullRef{aPull}, Named: named(checked)}, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) if err != nil { t.Fatalf("a failed copy failed the walk: %v", err) @@ -247,7 +293,7 @@ func TestAFailedCopyOfAChecksBuildFallsBackToBuilding(t *testing.T) { func TestOnlyAWalkOfATrunkCommitTakesAChecksBuild(t *testing.T) { withVersion(t, "builder-1") scratch := newScratch() - aCheckBuilt(t, scratch, "tree-one") + checked := aCheckBuilt(t, scratch, "tree-one") r, run, workspace := aModuleAt(t, "tree-one") if _, err := Build(t.Context(), run, r, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, @@ -259,7 +305,7 @@ func TestOnlyAWalkOfATrunkCommitTakesAChecksBuild(t *testing.T) { } r, _, workspace = aModuleAt(t, "tree-one") // no trunk said: off it - got, err := BuildReusing(t.Context(), r.run, r, Reuse{Scratch: scratch}, "https://forge.invalid/meshboard.git", + got, err := BuildReusing(t.Context(), r.run, r, Reuse{Scratch: scratch, Named: named(checked)}, "https://forge.invalid/meshboard.git", "mod", "main", workspace, nil, Npmrc{}, GitCredential{}, nil) if err != nil { t.Fatal(err) @@ -539,3 +585,43 @@ func TestAChecksBuildIsCopiedByDigestIntoTheModulesNamespace(t *testing.T) { t.Fatalf("the archive was copied as %+v", got) } } + +// Two checks that built one digest in one repository share it: letting go of one check's builds keeps it while the +// other still names it, and the last to go takes it. +func TestABuildTwoChecksShareGoesWithTheLastOfThem(t *testing.T) { + m := &aTaggingRegistry{aRegistryOfRepositories: newRegistryOfRepositories(), tags: map[string]map[string]string{}} + server := httptest.NewServer(m.serve()) + defer server.Close() + address := strings.TrimPrefix(server.URL, "http://") + registry := Registry{Address: address} + ctx := t.Context() + now := time.Date(2026, 10, 13, 20, 0, 0, 0, time.UTC) + + body := `{"mediaType":"` + mediaManifestOCI + `","layers":[]}` + digest := digestOf([]byte(body)) + m.manifests["scratch/meshboard/server@"+digest] = []byte(body) + reference := address + "/scratch/meshboard/server@" + digest + other := PullRef{Owner: "novox", Repo: "meshboard", Number: 8} + for i, pull := range []PullRef{aPull, other} { + id := "build-" + strconv.Itoa(i+1) + b := CheckBuild{Key: "r1:" + strings.Repeat(strconv.Itoa(i+1), 64), Module: "meshboard", Pull: pull, Check: id, + At: now, Made: []catalogue.Built{{Name: "server", Kind: "image", Reference: reference}}} + if err := registry.Keep(ctx, pull, id, now, []CheckBuild{b}); err != nil { + t.Fatal(err) + } + } + held := func() bool { _, ok := m.manifests["scratch/meshboard/server@"+digest]; return ok } + + if n, err := registry.Forget(ctx, aPull); err != nil || n != 1 { + t.Fatalf("forgot %d (%v)", n, err) + } + if !held() { + t.Fatal("letting go of one check's builds took a build another check still names") + } + if n, err := registry.Forget(ctx, other); err != nil || n != 1 { + t.Fatalf("forgot %d (%v)", n, err) + } + if held() { + t.Fatal("the build stayed after the last check naming it went") + } +} diff --git a/internal/link/build.go b/internal/link/build.go index 72640a2a..6eb16588 100644 --- a/internal/link/build.go +++ b/internal/link/build.go @@ -101,8 +101,22 @@ type BuildRequest struct { } // ReuseAsk is a walk's leave to take a check's build (novox/hq ADR 0350 §2). +// +// **Only a build a verdict names is taken.** Builds are the builds the build seat said, in the outcomes of its checks +// of the walk's own pull requests, that it made for their merges: each key with its artifacts' digests. The record +// in the scratch namespace is not trusted alone — the artifact store takes writes from anything on the network, a +// check runs a pull request's own script on that network, and a record planted under another merge's key would +// otherwise be registered. The builder takes a scratch build only when a build here names its key and every digest. type ReuseAsk struct { - Pulls []PullRef `json:"pulls,omitempty"` + Pulls []PullRef `json:"pulls,omitempty"` + Builds []CheckedBuild `json:"builds,omitempty"` +} + +// CheckedBuild is one build a check's verdict names: the pull request it checked, its ask, and the build. +type CheckedBuild struct { + Pull PullRef `json:"pull"` + Check string `json:"check"` + Built CheckBuilt `json:"built"` } // PullRef names a pull request. diff --git a/internal/link/check_builds_test.go b/internal/link/check_builds_test.go new file mode 100644 index 00000000..96ccbbf4 --- /dev/null +++ b/internal/link/check_builds_test.go @@ -0,0 +1,39 @@ +package link + +import ( + "encoding/json" + "testing" + "time" +) + +// The builds a walk may take are read from the build seat's own outcomes of its checks (novox/hq ADR 0350 §2): only +// checks of the walk's pull requests, only builds made, newest first. +func TestNatsTheBuildsAWalkMayTakeAreTheChecksOutcomesOfItsPullRequests(t *testing.T) { + js := aBus(t) + since := time.Now().Add(-time.Minute) + say := func(r BuildResult) { + body, _ := json.Marshal(r) + if _, err := js.Context().Publish(BuildOutcomeOf(TheBuildMachine), body); err != nil { + t.Fatal(err) + } + } + mine := &CheckRequest{Owner: "novox", Repo: "shop", Number: 7} + built := CheckBuilt{Module: "app", Key: "r1:1", Made: []MadeArtifact{{Name: "server", Reference: "s/app/server@sha256:a"}}} + say(BuildResult{ID: "build-1", Checked: mine, Check: &CheckOutcome{Verdict: "pass", Built: []CheckBuilt{built, + {Module: "web", NotBuilt: "it has no reuse key"}}}}) + say(BuildResult{ID: "build-2", Checked: &CheckRequest{Owner: "novox", Repo: "shop", Number: 8}, + Check: &CheckOutcome{Verdict: "pass", Built: []CheckBuilt{built}}}) + say(BuildResult{ID: "build-3", Repository: "novox/shop", Commit: "c0ffee"}) // a build, not a check + newer := built + newer.Key = "r1:2" + say(BuildResult{ID: "build-4", Checked: mine, Check: &CheckOutcome{Verdict: "pass", Built: []CheckBuilt{newer}}}) + + got, err := ReadCheckBuilds(t.Context(), js, TheBuildMachine, []PullRef{{Owner: "novox", Repo: "shop", Number: 7}}, since) + if err != nil { + t.Fatal(err) + } + if len(got) != 2 || got[0].Check != "build-4" || got[0].Built.Key != "r1:2" || got[1].Check != "build-1" || + got[1].Pull.Number != 7 { + t.Fatalf("read %+v", got) + } +} diff --git a/internal/link/queue.go b/internal/link/queue.go index b253909e..50865ef8 100644 --- a/internal/link/queue.go +++ b/internal/link/queue.go @@ -364,6 +364,64 @@ func ReadBuildEvents(ctx context.Context, js *broker.JetStream, seat string, sin return heard, nil } +// ReadCheckBuilds is every build the seat's checks of those pull requests said they made for their merges since a +// moment (novox/hq ADR 0350 §2): read from the seat's own outcomes on the events stream, which only its holders may +// publish, never from the artifact store. Newest first. +func ReadCheckBuilds(ctx context.Context, js *broker.JetStream, seat string, pulls []PullRef, since time.Time) ([]CheckedBuild, error) { + if len(pulls) == 0 { + return nil, nil + } + wanted := map[PullRef]bool{} + for _, p := range pulls { + wanted[p] = true + } + subject := BuildOutcomeOf(seat) + sub, err := js.Context().PullSubscribe(subject, "", + nats.BindStream(broker.EventsStream), nats.StartTime(since), nats.AckNone()) + if err != nil { + return nil, fmt.Errorf("cannot read %s from the bus: %w", subject, err) + } + defer func() { _ = sub.Unsubscribe() }() + info, err := sub.ConsumerInfo() + if err != nil { + return nil, fmt.Errorf("cannot read %s from the bus: %w", subject, err) + } + var out []CheckedBuild + for remaining := info.NumPending; remaining > 0; { + batch, err := sub.Fetch(int(min(remaining, 500)), nats.MaxWait(2*time.Second)) + if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) { + return nil, fmt.Errorf("reading %s: %w", subject, err) + } + if len(batch) == 0 { + break + } + remaining -= uint64(len(batch)) + for _, msg := range batch { + var r BuildResult + if json.Unmarshal(msg.Data, &r) != nil || r.Checked == nil || r.Check == nil || r.Failed != "" { + continue + } + pull := PullRef{Owner: r.Checked.Owner, Repo: r.Checked.Repo, Number: r.Checked.Number} + if !wanted[pull] { + continue + } + for _, b := range r.Check.Built { + if b.NotBuilt == "" && b.Key != "" { + out = append(out, CheckedBuild{Pull: pull, Check: r.ID, Built: b}) + } + } + } + if ctx.Err() != nil { + return nil, ctx.Err() + } + } + // Newest first: the stream's order is the order they were said. + for i, j := 0, len(out)-1; i < j; i, j = i+1, j-1 { + out[i], out[j] = out[j], out[i] + } + return out, nil +} + // --- the cancelled set ---------------------------------------------------------------------- // safeKey is an id that can be a key, and a subject token, as it is: a build id, never a pattern.