From 5091cab65274460157b8040cc433b13a15c2b0ee Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:27:59 +0200 Subject: [PATCH] Read the assigner's composition into the checker's verdict in place of 'not yet run', failing a check whose composition failed, naming its machine (issue 1297, subtask B3d; ADR 0528 clause 5) --- .../cmd/mesh-delivery-checker/compositions.go | 45 ++++++++++ .../compositions_integration_test.go | 82 +++++++++++++++++++ .../cmd/mesh-delivery-checker/judge.go | 6 +- .../cmd/mesh-delivery-checker/main.go | 13 ++- modules/mesh-delivery-checker/go.mod | 2 +- .../internal/check/composition_test.go | 80 ++++++++++++++++++ .../internal/check/judge.go | 43 +++++++++- .../mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 1 + 9 files changed, 266 insertions(+), 8 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go create mode 100644 modules/mesh-delivery-checker/internal/check/composition_test.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go new file mode 100644 index 0000000..951b0a6 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -0,0 +1,45 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key: no poll and no +// timer, the check's own bound ends the wait. +type compositionWatch struct{ kv jetstream.KeyValue } + +// Await answers the composition under key once the assigner wrote one that does not wait; a value it cannot read is an +// error, and ctx ending is ctx's error. +func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { + w, err := c.kv.Watch(ctx, key) + if err != nil { + return delivery.CompositionVerdict{}, err + } + defer w.Stop() + for { + select { + case <-ctx.Done(): + return delivery.CompositionVerdict{}, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + v, err := delivery.DecodeCompositionVerdict(e.Value()) + if err != nil { + return delivery.CompositionVerdict{}, fmt.Errorf("the composition %s cannot be read: %w", key, err) + } + if v.State != delivery.CompositionWaits { + return v, nil + } + } + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go new file mode 100644 index 0000000..070abae --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -0,0 +1,82 @@ +//go:build integration + +package main + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is +// absent and while the assigner says it waits, and answers the verdict the assigner then writes under the check's key. +func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "comp") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.CompositionKey("feat/x", "0123456789ab") + got := make(chan delivery.CompositionVerdict, 1) + go func() { + v, err := compositionWatch{kv}.Await(ctx, key) + if err != nil { + t.Error(err) + } + got <- v + }() + put := func(v delivery.CompositionVerdict) { + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, data); err != nil { + t.Fatal(err) + } + } + time.Sleep(200 * time.Millisecond) + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "the indexer holds no candidates yet"}) + select { + case v := <-got: + t.Fatalf("a waiting composition was answered: %+v", v) + case <-time.After(300 * time.Millisecond): + } + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}}) + select { + case v := <-got: + if v.State != delivery.CompositionPass || v.Machines[0] != "g14" { + t.Errorf("answered %+v", v) + } + case <-time.After(5 * time.Second): + t.Fatal("the composition written was not answered") + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index ed8ff96..ecc4e6b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -27,11 +27,13 @@ type judgeDeps struct { forge check.Forge announce check.Announcer builds check.BuildStates - logf func(format string, args ...any) + // compositions is the assigner's bucket of compositions; nil leaves every composition not run. + compositions check.CompositionReader + logf func(format string, args ...any) } func newJudge(d judgeDeps) check.Judge { - return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, + return &check.Examiner{Compositions: d.compositions, Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 41377e8..676d380 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -222,6 +222,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro default: builds = buildReader{kvStore{kv}} } + // The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the + // composition was not run, until the checker starts again. + var compositions check.CompositionReader + switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) + default: + compositions = compositionWatch{kv} + } // The tries (ADR 0501 ยง5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger) @@ -238,7 +249,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf, - announce: announce, + announce: announce, compositions: compositions, }) ch.Announce = announce tryer.Judge = ch.Judge diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 0de4a55..e42ae23 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/internal/check/composition_test.go b/modules/mesh-delivery-checker/internal/check/composition_test.go new file mode 100644 index 0000000..19bb6b6 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/composition_test.go @@ -0,0 +1,80 @@ +package check + +import ( + "context" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeCompositions answers the assigner's composition of a check from memory, recording the keys asked. +type fakeCompositions struct { + v delivery.CompositionVerdict + err error + asked []string +} + +func (f *fakeCompositions) Await(_ context.Context, key string) (delivery.CompositionVerdict, error) { + f.asked = append(f.asked, key) + return f.v, f.err +} + +// 1297/AC3 (its unit half): a check whose tests pass takes the assigner's composition of it, by the check's own key, +// into its verdict: a pass says it passed with the mesh-sdk and machines it composed, and is no longer "not run". +func TestAPassingCheckTakesTheAssignersComposition(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "v0.1.14-x", Machines: []string{"g14"}, Took: 40}} + b.ex.Compositions = f + cs := caseOf(t, pullOf(repoA, 7, head1)) + v := b.judge(t, cs).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionPass || v.Composition.SDK != "v0.1.14-x" || v.Took.Composition != 40 { + t.Fatalf("verdict = %+v, composition %+v", v, v.Composition) + } + if len(f.asked) != 1 || f.asked[0] != cs.Key { + t.Errorf("asked %v, want the check's key %s", f.asked, cs.Key) + } + if !compositionMentions(v, "passed") { + t.Errorf("the line says %q", compositionWords(v.Composition, v.Took.Composition)) + } +} + +func compositionMentions(v delivery.Verdict, word string) bool { + return strings.Contains(compositionWords(v.Composition, v.Took.Composition), word) +} + +// 1297/AC2 at the checker: a failed composition fails the check, naming the machine it failed at and its words. +func TestAFailedCompositionFailsTheCheck(t *testing.T) { + b := newBench(t) + b.ex.Compositions = &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionFail, SDK: "x", + Machine: "g14", Words: "g14 runs web, which the change removes from its source"}} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictFail || v.Failed.Machine != "g14" || v.Composition == nil || v.Composition.State != delivery.CompositionFail { + t.Fatalf("verdict = %+v", v) + } + v.Group, v.Heads = "feat/x", "0123456789ab" // the checker names them when it writes the verdict + if _, err := delivery.EncodeVerdict(v); err != nil { + t.Errorf("the verdict cannot be written: %v", err) + } +} + +// 1297/B3d: a composition that cannot be read leaves it not run; a try is not composed. +func TestACompositionNotHadIsNotRun(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{err: errors.New("the check's bound passed")} + b.ex.Compositions = f + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionNotRun { + t.Fatalf("verdict = %+v", v) + } + b2 := newBench(t) + f2 := &fakeCompositions{v: delivery.CompositionVerdict{State: delivery.CompositionFail, Words: "w", SDK: "x"}} + b2.ex.Compositions = f2 + cs := caseOf(t, pullOf(repoA, 7, head1)) + cs.Try = &TryOptions{} + b2.judge(t, cs) + if len(f2.asked) != 0 { + t.Errorf("a try asked for a composition: %v", f2.asked) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index dcd07f0..099caca 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -105,8 +105,18 @@ type Examiner struct { } // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } - Now func() time.Time - Logf func(format string, args ...any) + // Compositions answers the assigner's composition check of a check by the check's key (delivery.CompositionKey), + // waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil + // leaves the composition not run. + Compositions CompositionReader + Now func() time.Time + Logf func(format string, args ...any) +} + +// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote +// a verdict that does not wait, or ctx ends. +type CompositionReader interface { + Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) } func (j *Examiner) now() time.Time { @@ -230,13 +240,40 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) { return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil } -// finish is the outcome of the verdict with the state of the builds the candidate's modules carry. +// finish is the outcome of the verdict with the state of the builds the candidate's modules carry, and, for a check +// whose tests passed, the assigner's composition of it. func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { out := st.outcome(v) + if out.Verdict.State == delivery.VerdictPass { + st.composed(ctx, &out.Verdict) + } out.Builds = st.j.buildNotes(ctx, st.c) return out } +// composed reads the assigner's composition of the check into its verdict (ADR 0528 clause 5): a failed composition +// fails the check, naming the machine it failed at. A try is not composed (it has no candidate); a composition that +// cannot be had leaves it not run, said. +func (st *state) composed(ctx context.Context, v *delivery.Verdict) { + if st.j.Compositions == nil || st.c.Try != nil { + return + } + comp, err := st.j.Compositions.Await(ctx, st.c.Key) + if err != nil { + st.j.logf("%s: the composition could not be read, so it is said as not run: %v", st.c.Key, err) + return + } + v.Composition = comp.Composition() + v.Took.Composition = comp.Took + if comp.State == delivery.CompositionFail { + machine := comp.Machine + if machine == "" { + machine = "the composition" + } + v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words} + } +} + // outcome fills the verdict's levels, times, machines and composition from what ran. func (st *state) outcome(v delivery.Verdict) Outcome { if v.State == delivery.VerdictPass || len(v.Levels) == 0 { diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index a4b5442..0fc6074 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", + "mesh-controller.plans-building", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 2c22576..1f15b6c 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -22,6 +22,7 @@ "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.plans-building", + "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",