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.