novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

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)

Browse Source
This commit is contained in:
jschoubben
2026-10-23 15:27:59 +02:00
parent d8599855e1
commit 5091cab652
9 changed files with 266 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go
+45
View File
@@ -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
}
}
}
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go
+82
View File
@@ -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")
}
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go
+4 -2
View File
@@ -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}
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+12 -1
View File
@@ -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
modules/mesh-delivery-checker/go.mod
+1 -1
View File
@@ -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
)
modules/mesh-delivery-checker/internal/check/composition_test.go
+80
View File
@@ -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)
}
}
modules/mesh-delivery-checker/internal/check/judge.go
+40 -3
View File
@@ -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 {
modules/mesh-delivery-checker/manifest_test.go
+1 -1
View File
@@ -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)
}
modules/mesh-delivery-checker/module.json
+1
View File
@@ -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",