diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go deleted file mode 100644 index 1647482..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/composition_line_integration_test.go +++ /dev/null @@ -1,64 +0,0 @@ -//go:build integration - -package main - -import ( - "strings" - "testing" - - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/delivery" -) - -// assignerWrote raises the assigner's compositions bucket, the assigner online in the watchman's presence, and its verdict -// under the check's key, as the assigner leaves them on the bus. -func (b *bed) assignerWrote(t *testing.T, key string, v delivery.CompositionVerdict) { - t.Helper() - kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory}) - if err != nil { - t.Fatal(err) - } - pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox") - on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) - if _, err := b.buckets[delivery.PresenceBucket].Put(b.ctx, pkey, on); err != nil { - t.Fatal(err) - } - data, err := delivery.EncodeCompositionVerdict(v) - if err != nil { - t.Fatal(err) - } - if _, err := kv.Put(b.ctx, key, data); err != nil { - t.Fatal(err) - } -} - -// 1297/AC3: started whole with the assigner's verdict on the bus, the checker's line says the composition's verdict, not -// "not yet run": a pass names its machine, a fail fails the check with the assigner's words. -func TestTheCheckersLineSaysTheAssignersComposition(t *testing.T) { - t.Run("the read awaits the assigner's composition", checkerAwaitsTheAssignersComposition) - for _, c := range []struct { - state, owed, words string - }{ - {delivery.CompositionPass, "success", "composition passed"}, - {delivery.CompositionFail, "failure", "g14 runs web, which the change removes from its source"}, - } { - b := startBed(t) - key := b.candidate(t, "feat/x", pull(repo, 7, head1)) - group, heads, _ := delivery.ParseCheckerVerdictKey(key) - v := delivery.CompositionVerdict{Group: group, Heads: heads, State: c.state, SDK: "v0.1.14-x", Machines: []string{"g14"}} - if c.state == delivery.CompositionFail { - v.Machine, v.Words = "g14", c.words - } - b.assignerWrote(t, key, v) - b.start(t, nil) - got := b.verdict(t, key, ended) - if got.Composition == nil || got.Composition.State != c.state { - t.Fatalf("%s: the verdict's composition is %+v", c.state, got.Composition) - } - o := b.owed(t, 7, head1, c.owed) - if !strings.Contains(o.Text, c.words) || strings.Contains(o.Text, "not yet run") { - t.Errorf("%s: the line says %q", c.state, o.Text) - } - } -} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go deleted file mode 100644 index c5a3fe3..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go +++ /dev/null @@ -1,131 +0,0 @@ -package main - -import ( - "context" - "errors" - "sync" - - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-checker/internal/check" -) - -// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key and the -// assigner's presence: no poll and no timer. With no assigner online it answers check.ErrAssignerNotRunning at once. -type compositionWatch struct { - kv jetstream.KeyValue - // presence is the watchman's presence bucket; nil (not there) counts as no assigner online. - presence jetstream.KeyValue -} - -// Await answers the composition under key once the assigner wrote one that does not wait (check.AwaitComposition). -func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { - ctx, cancel := context.WithCancel(ctx) - defer cancel() - comps := make(chan check.CompositionEvent) - presence := make(chan check.PresenceEvent) - cw, err := c.kv.Watch(ctx, key) - if err != nil { - return delivery.CompositionVerdict{}, err - } - defer cw.Stop() - go func() { - for e := range cw.Updates() { - ev := check.CompositionEvent{Settled: e == nil} - if e != nil && e.Operation() == jetstream.KeyValuePut { - ev.Value = e.Value() - } - select { - case comps <- ev: - case <-ctx.Done(): - return - } - } - }() - if c.presence == nil { - go func() { - select { - case presence <- check.PresenceEvent{Settled: true}: - case <-ctx.Done(): - } - }() - } else { - pw, err := c.presence.Watch(ctx, "part."+delivery.AssignerModule+".*") - if err != nil { - return delivery.CompositionVerdict{}, err - } - defer pw.Stop() - go func() { - for e := range pw.Updates() { - ev := check.PresenceEvent{Settled: e == nil} - if e != nil { - ev.Key = e.Key() - if e.Operation() == jetstream.KeyValuePut { - if p, err := delivery.DecodePresence(e.Value()); err == nil { - ev.Online = p.Online - } - } - } - select { - case presence <- ev: - case <-ctx.Done(): - return - } - } - }() - } - return check.AwaitComposition(ctx, presence, comps) -} - -// followCompositions hands the checker each composition the assigner writes from now on, so a verdict that said its -// composition was not judged takes it in when the assigner starts (event-driven). -func followCompositions(ctx context.Context, kv jetstream.KeyValue, ch *check.Checker) error { - w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly()) - if err != nil { - return err - } - go func() { - defer w.Stop() - for e := range w.Updates() { - if e == nil || e.Operation() != jetstream.KeyValuePut { - continue - } - if comp, err := delivery.DecodeCompositionVerdict(e.Value()); err == nil { - ch.Composed(ctx, comp) - } - } - }() - return nil -} - -// lazyCompositions opens the assigner's compositions when a check first asks for one and the bucket is there: until it -// is, no assigner ever ran, so the check says at once that its composition is not judged (check.ErrAssignerNotRunning), -// and the next check looks again. No timer: each look is a check asking. -type lazyCompositions struct { - open func(ctx context.Context) (check.CompositionReader, error) - - mu sync.Mutex - reader check.CompositionReader -} - -// Await is the opened reader's Await, opening it first when it is not yet. -func (l *lazyCompositions) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { - l.mu.Lock() - r := l.reader - if r == nil { - opened, err := l.open(ctx) - switch { - case errors.Is(err, jetstream.ErrBucketNotFound): - l.mu.Unlock() - return delivery.CompositionVerdict{}, check.ErrAssignerNotRunning - case err != nil: - l.mu.Unlock() - return delivery.CompositionVerdict{}, err - } - l.reader, r = opened, opened - } - l.mu.Unlock() - return r.Await(ctx, key) -} 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 deleted file mode 100644 index 0c147e7..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go +++ /dev/null @@ -1,107 +0,0 @@ -//go:build integration - -package main - -import ( - "context" - "errors" - "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" - - "mesh-delivery-checker/internal/check" -) - -// 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. -// A case of TestTheCheckersLineSaysTheAssignersComposition (merged into it for the test budget, issue 774). -func checkerAwaitsTheAssignersComposition(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) - } - pkv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.PresenceBucket, History: 5}) - if err != nil { - t.Fatal(err) - } - pkey, _ := delivery.PresencePartKey(delivery.AssignerModule, "novox") - on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) - if _, err := pkv.Put(ctx, pkey, on); err != nil { - t.Fatal(err) - } - // With no assigner online the wait answers at once (a second key, before the assigner of this test is online). - off, _ := delivery.EncodePresence(delivery.Presence{Online: false}) - if _, err := pkv.Put(ctx, pkey, off); err != nil { - t.Fatal(err) - } - other, _ := delivery.CompositionKey("feat/y", "0123456789ab") - if _, err := (compositionWatch{kv: kv, presence: pkv}).Await(ctx, other); !errors.Is(err, check.ErrAssignerNotRunning) { - t.Fatalf("with no assigner online the wait answered %v", err) - } - if _, err := pkv.Put(ctx, pkey, on); err != nil { - t.Fatal(err) - } - key, _ := delivery.CompositionKey("feat/x", "0123456789ab") - got := make(chan delivery.CompositionVerdict, 1) - go func() { - v, err := compositionWatch{kv: kv, presence: pkv}.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/compositions_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go deleted file mode 100644 index 64afd0b..0000000 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_test.go +++ /dev/null @@ -1,46 +0,0 @@ -package main - -import ( - "context" - "errors" - "testing" - - "github.com/nats-io/nats.go/jetstream" - - "git.novox.be/novox/mesh-sdk/go/delivery" - - "mesh-delivery-checker/internal/check" -) - -type answering struct{ v delivery.CompositionVerdict } - -func (a answering) Await(context.Context, string) (delivery.CompositionVerdict, error) { - return a.v, nil -} - -// 1297 review item 3: the compositions bucket missing when the checker starts is looked up again at the next check: while -// it is missing a check says at once the assigner is not running; once it is there the check reads its composition, and -// it is opened once. -func TestTheCompositionsBucketIsLookedUpAtEachCheck(t *testing.T) { - opens, there := 0, false - l := &lazyCompositions{open: func(context.Context) (check.CompositionReader, error) { - opens++ - if !there { - return nil, jetstream.ErrBucketNotFound - } - return answering{delivery.CompositionVerdict{State: delivery.CompositionPass, SDK: "x"}}, nil - }} - if _, err := l.Await(context.Background(), "k"); !errors.Is(err, check.ErrAssignerNotRunning) { - t.Fatalf("without the bucket: %v", err) - } - there = true - for range 2 { - v, err := l.Await(context.Background(), "k") - if err != nil || v.State != delivery.CompositionPass { - t.Fatalf("with the bucket: %+v, %v", v, err) - } - } - if opens != 2 { - t.Errorf("opened %d times, want twice (missing, then once)", opens) - } -} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/engines.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/engines.go new file mode 100644 index 0000000..bc193f5 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/engines.go @@ -0,0 +1,140 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "sync" + "time" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun" + "git.novox.be/novox/mesh-sdk/go/buildstep" + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/composer" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The checker's composition of a check on the bus (novox/hq ADR 0593 §7, §9; check.EngineCompositions): each placed +// node's engine asked `node-engine.compose-candidate` on that node, and, for a group that changes the node-engine's +// composition code, the composer built from its head run on the Go build seat. + +// composeCandidateKey is the seat verb's call key, asked of one node (process.Conn.CallOn). +var composeCandidateKey = "seat:" + catalogue.EngineSeat + "." + delivery.EngineComposeCandidateVerb + +// engineAsks asks one node's engine over the bus, its answer waited for at most ctx's deadline (the answer bound). +type engineAsks struct { + call func(node, key string, body any) (json.RawMessage, error) +} + +func (e engineAsks) ComposeCandidate(ctx context.Context, node string, ask delivery.ComposeCandidateAsk) (delivery.ComposeCandidateAnswer, error) { + type answer struct { + raw json.RawMessage + err error + } + got := make(chan answer, 1) + go func() { + raw, err := e.call(node, composeCandidateKey, ask) + got <- answer{raw, err} + }() + select { + case <-ctx.Done(): + return delivery.ComposeCandidateAnswer{}, fmt.Errorf("no answer within its bound: %w", ctx.Err()) + case a := <-got: + if a.err != nil { + return delivery.ComposeCandidateAnswer{}, a.err + } + return delivery.DecodeComposeCandidateAnswer(a.raw) + } +} + +// placedNodes names every node the seat mesh-placement places now, read from its state when a check composes (no +// watch: a check reads it once). A state that cannot be read places no node, said in the journal. +type placedNodes struct { + kv func(ctx context.Context) (jetstream.KeyValue, error) + logf func(string, ...any) +} + +func (p placedNodes) Placed() []string { + ctx, cancel := context.WithTimeout(context.Background(), delivery.ComposeAnswerBound) + defer cancel() + kv, err := p.kv(ctx) + if err != nil { + p.logf("the state %s cannot be read, so no node is asked to compose: %v", delivery.PlacementBucket, err) + return nil + } + lister, err := kv.ListKeys(ctx) + if err != nil { + p.logf("the state %s cannot be listed, so no node is asked to compose: %v", delivery.PlacementBucket, err) + return nil + } + seen := map[string]bool{} + for key := range lister.Keys() { + if node, _, _, err := delivery.ParsePlacementKey(key); err == nil { + seen[node] = true + } + } + out := make([]string, 0, len(seen)) + for n := range seen { + out = append(out, n) + } + sort.Strings(out) + return out +} + +// seatSandbox runs a compose job on the Go build seat's check stream, as the checker's own ask, and answers the +// composer's verdict from the step's result (delivery.ComposeAnswer). +type seatSandbox struct{ asker buildrun.Asker } + +func (s seatSandbox) Compose(ctx context.Context, job delivery.ComposeJob) (composer.Verdict, error) { + ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: delivery.AskCheck, Language: delivery.AskGo, Asker: delivery.CheckerModule, + At: time.Now().UTC(), Compose: &job} + step, err := s.asker.Ask(ctx, ask) + if err != nil { + return composer.Verdict{}, err + } + if step.State != delivery.StepPassed { + return composer.Verdict{}, fmt.Errorf("the compose ask %s ended %s: %s", ask.ID, step.State, step.Reason) + } + var a delivery.ComposeAnswer + if err := json.Unmarshal(step.Result, &a); err != nil { + return composer.Verdict{}, fmt.Errorf("the compose ask %s answered no compose answer: %w", ask.ID, err) + } + var v composer.Verdict + if err := json.Unmarshal(a.Verdict, &v); err != nil || v.V != composer.Version { + return composer.Verdict{}, fmt.Errorf("the compose ask %s answered no composer verdict of version %d (%v)", ask.ID, composer.Version, err) + } + return v, nil +} + +// manifestVerdicts are the indexer's refusals of the trunk's manifests, by module, from its modules' values +// (delivery.Module.VerdictOf, ADR 0578 §3): what the sandbox's composer holds a refused module by. +type manifestVerdicts struct { + mu sync.Mutex + by map[string]delivery.ManifestVerdict +} + +func (m *manifestVerdicts) put(module string, v *delivery.ManifestVerdict) { + m.mu.Lock() + defer m.mu.Unlock() + if m.by == nil { + m.by = map[string]delivery.ManifestVerdict{} + } + if v == nil { + delete(m.by, module) + return + } + m.by[module] = *v +} + +func (m *manifestVerdicts) all() map[string]delivery.ManifestVerdict { + m.mu.Lock() + defer m.mu.Unlock() + out := make(map[string]delivery.ManifestVerdict, len(m.by)) + for k, v := range m.by { + out[k] = v + } + return out +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go index c1018c8..3a65048 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go @@ -10,6 +10,7 @@ import ( "io" "os" "path/filepath" + "slices" "sort" "strings" "sync" @@ -262,6 +263,58 @@ type bed struct { mu sync.Mutex } +// engine places each node and serves compose-candidate for the first one, as its node-engine does on the bus (answered on +// the seat verb's caller-named and plain subjects); the others never answer. It answers what it was asked. +func (b *bed) engine(t *testing.T, nodes ...string) func() []delivery.ComposeCandidateAsk { + t.Helper() + kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: 1}) + if err != nil { + t.Fatal(err) + } + for _, n := range nodes { + pk, _ := delivery.PlacementNodeKey(n) + raw, _ := json.Marshal(delivery.Placement{Node: n}) + if _, err := kv.Put(b.ctx, pk, raw); err != nil { + t.Fatal(err) + } + } + nc, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + var mu sync.Mutex + var asked []delivery.ComposeCandidateAsk + answer := func(m *nats.Msg) { + ask, err := delivery.DecodeComposeCandidateAsk(m.Data) + if err != nil { + _ = m.Respond([]byte(`{"error":"` + err.Error() + `"}`)) + return + } + mu.Lock() + asked = append(asked, ask) + mu.Unlock() + a := delivery.ComposeCandidateAnswer{Node: nodes[0], Candidate: ask.Candidate, Verdict: delivery.ComposeCandidateComposed, + Own: delivery.ModuleComposed{Module: "node", State: delivery.ModuleComposedState, Digest: "sha256:" + strings.Repeat("b", 64)}} + raw, err := delivery.EncodeComposeCandidateAnswer(a) + if err != nil { + t.Error(err) + } + out, _ := json.Marshal(map[string]any{"result": json.RawMessage(raw), "node": nodes[0]}) + _ = m.Respond(out) + } + for _, subject := range []string{"mesh.seat.node-engine.call.compose-candidate." + nodes[0] + ".>", "mesh.seat.node-engine.tool.compose-candidate." + nodes[0]} { + if _, err := nc.Subscribe(subject, answer); err != nil { + t.Fatal(err) + } + } + return func() []delivery.ComposeCandidateAsk { + mu.Lock() + defer mu.Unlock() + return slices.Clone(asked) + } +} + // startBed starts the bus, the buckets the controller makes from the checker's and the other parts' manifests, the build // modules' streams and the checker's durable consumers on them, one node's toolchains, the gate's source naming the trunk, and // the fake build modules. It runs no checker. @@ -509,14 +562,20 @@ func (b *bed) owed(t *testing.T, number int, head, state string) delivery.Owed { func TestCheckerJudgesACandidateAndOwesItsStatus(t *testing.T) { // N6/AC5, 1300/AC5, 1445/AC1, 1929/AC1 b := startBed(t) key := b.candidate(t, "feat/x", pull(repo, 7, head1)) + // ADR 0593 §7, §9 (D12): n1 and n2 are placed; n1's engine answers compose-candidate on the bus, n2's never does. + asked := b.engine(t, "n1", "n2") b.start(t, nil) v := b.verdict(t, key, ended) if v.State != delivery.VerdictPass || v.Levels[delivery.LevelUnit].Run != 2 || v.Levels[delivery.LevelIntegration].Run != 2 || v.Trunk[repo] != trunk { t.Fatalf("verdict = %+v", v) } - if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || len(v.Machines) != 1 || v.Machines[0] != "n1" { + if c := v.Composition; c == nil || c.State != delivery.CompositionPass || !slices.Equal(c.Machines, []string{"n1"}) || + !strings.Contains(c.Words, "n2: "+delivery.ComposeNoAnswer) || len(v.Machines) != 1 || v.Machines[0] != "n1" { t.Errorf("composition %+v machines %v", v.Composition, v.Machines) } + if got := asked(); len(got) == 0 || got[0].Candidate != key { + t.Errorf("n1's engine was asked %+v, want the check's candidate %s", got, key) + } b.fake.mu.Lock() names := b.fake.names b.fake.mu.Unlock() 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 e4589b4..c1fc816 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -31,7 +31,7 @@ type judgeDeps struct { forge check.Forge announce check.Announcer builds check.BuildStates - // compositions is the assigner's bucket of compositions; nil leaves every composition not run. + // compositions makes a check's composition from the engines' answers (ADR 0593 §7); nil leaves it not run. compositions check.CompositionReader // nodes names every node with a record; nil hands no names. nodes *nodeNames @@ -111,6 +111,8 @@ type manifestKV interface { type repoModules struct { manifests manifestKV logf func(format string, args ...any) + // verdicts keeps the indexer's verdict on each module's manifest, for the sandbox's composer (engines.go); nil keeps none. + verdicts *manifestVerdicts mu sync.Mutex langs map[string]map[string]bool // by manifest digest: the languages of its artifacts @@ -130,6 +132,9 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted r.tsPkgs, r.path, r.named = map[string][]string{}, map[string]string{}, map[string]map[string][]string{} } if deleted { + if r.verdicts != nil { + r.verdicts.put(key, nil) + } delete(r.repoOf, key) delete(r.typed, key) delete(r.commit, key) @@ -143,6 +148,9 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted if err != nil { return } + if r.verdicts != nil { + r.verdicts.put(key, m.VerdictOf(key)) + } r.mu.Lock() r.repoOf[key] = m.Repository r.typed[key] = m.Manifest 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 3296872..1f03586 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -169,23 +169,20 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if kv := process.OptionalBucket(ctx, js, delivery.BuilderBuildsBucket, "the line says nothing of the builds until the checker starts again", logger.Printf); kv != nil { builds = buildReader{kvStore{kv}} } - // The assigner's compositions (ADR 0528 clause 5, issue 1297): their bucket is looked up at each check until it is - // there, so an assigner delivered after the checker started is read at the next check, and followed from then on. - compositions := &lazyCompositions{open: func(at context.Context) (check.CompositionReader, error) { - kv, err := js.KeyValue(at, delivery.AssignerCompositionsBucket) - if err != nil { - return nil, err - } - cw := compositionWatch{kv: kv} - if pkv, err := js.KeyValue(at, delivery.PresenceBucket); err == nil { - cw.presence = pkv - } - if err := followCompositions(ctx, kv, ch); err != nil { - return nil, fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err) - } - logger.Printf("the assigner's compositions are read from %s", delivery.AssignerCompositionsBucket) - return cw, nil - }} + // The check's composition is the checker's, from what each placed node's engine answers (ADR 0593 §7, §9). + answerBound, err := composeAnswerBound(getenv("MESH_CHECKER_SETTINGS")) + if err != nil { + return err + } + verdicts := &manifestVerdicts{} + compositions := check.EngineCompositions{ + Asker: engineAsks{call: caller.CallOn}, + Placed: placedNodes{kv: func(at context.Context) (jetstream.KeyValue, error) { return js.KeyValue(at, delivery.PlacementBucket) }, + logf: logger.Printf}.Placed, + Bound: answerBound, + Sandbox: seatSandbox{asker: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}}, + Verdicts: verdicts.all, + } // 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, NotTaken: stepBound, Logf: logger.Printf} announce := func(ctx context.Context, event, key string) { @@ -194,7 +191,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("%s of %s was written and the part that reads it could not be told: %v", event, key, err) } } - repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf} + repos := &repoModules{manifests: kvStore{buckets[delivery.ManifestsBucket]}, logf: logger.Printf, verdicts: verdicts} + compositions.Composer = func() string { return repos.CommitOf(check.NodeEngineModule) } + compositions.Manifests = repos // The indexer's packages state (ADR 0593 §2): followed once the bus has it, so a Go check builds every mesh package at // its published version; until then a check builds each at the version its go.mod names. published := &publishedPackages{} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go new file mode 100644 index 0000000..c65f0f7 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings.go @@ -0,0 +1,44 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "io/fs" + "os" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// The checker's own settings (its config, written by the node-engine into ${dir:state}/settings.json, named to the +// process by MESH_CHECKER_SETTINGS): compose-answer-bound, the seconds a check waits for a node's engine to answer +// compose-candidate (novox/hq ADR 0593 §7, §9: 60 by default; the checker owns it). + +// composeAnswerBound reads compose-answer-bound from the settings file at path: delivery.ComposeAnswerBound where no +// file or no value is there; a value that is no number of seconds from 1 to 600 is refused, naming it. +func composeAnswerBound(path string) (time.Duration, error) { + if path == "" { + return delivery.ComposeAnswerBound, nil + } + raw, err := os.ReadFile(path) + if errors.Is(err, fs.ErrNotExist) { + return delivery.ComposeAnswerBound, nil + } + if err != nil { + return 0, err + } + var s struct { + Bound *float64 `json:"compose-answer-bound"` + } + if err := json.Unmarshal(raw, &s); err != nil { + return 0, fmt.Errorf("%s is no settings file: %w", path, err) + } + if s.Bound == nil { + return delivery.ComposeAnswerBound, nil + } + if *s.Bound < 1 || *s.Bound > 600 || *s.Bound != float64(int(*s.Bound)) { + return 0, fmt.Errorf("compose-answer-bound is %v: a whole number of seconds from 1 to 600", *s.Bound) + } + return time.Duration(*s.Bound) * time.Second, nil +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go new file mode 100644 index 0000000..37155a3 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/settings_test.go @@ -0,0 +1,35 @@ +package main + +import ( + "os" + "path/filepath" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ADR 0593 §7: the answer bound is the checker's setting compose-answer-bound, 60 s when the file or the value is +// absent; a value that is no whole number of seconds from 1 to 600 stops the start, naming it. +func TestTheAnswerBoundIsTheCheckersSetting(t *testing.T) { + dir := t.TempDir() + for _, row := range []struct { + body string + want time.Duration + err bool + }{{"", delivery.ComposeAnswerBound, false}, {`{}`, delivery.ComposeAnswerBound, false}, + {`{"compose-answer-bound": 90}`, 90 * time.Second, false}, {`{"compose-answer-bound": 0}`, 0, true}, + {`{"compose-answer-bound": 1.5}`, 0, true}} { + path := filepath.Join(dir, "settings.json") + os.Remove(path) + if row.body != "" { + if err := os.WriteFile(path, []byte(row.body), 0o600); err != nil { + t.Fatal(err) + } + } + got, err := composeAnswerBound(path) + if (err != nil) != row.err || got != row.want { + t.Errorf("%q: %s, %v", row.body, got, err) + } + } +} diff --git a/modules/mesh-delivery-checker/files/settings.json b/modules/mesh-delivery-checker/files/settings.json new file mode 100644 index 0000000..b9a1967 --- /dev/null +++ b/modules/mesh-delivery-checker/files/settings.json @@ -0,0 +1,3 @@ +{ + "compose-answer-bound": 60 +} diff --git a/modules/mesh-delivery-checker/internal/check/awaiting.go b/modules/mesh-delivery-checker/internal/check/awaiting.go deleted file mode 100644 index f2182b1..0000000 --- a/modules/mesh-delivery-checker/internal/check/awaiting.go +++ /dev/null @@ -1,168 +0,0 @@ -package check - -import ( - "context" - "errors" - "fmt" - "strings" - - "git.novox.be/novox/mesh-sdk/go/delivery" -) - -// ErrAssignerNotRunning is what a wait for a composition answers when no instance of the assigner is online: nobody is -// making it, so the check does not wait for it (issue 1297). The verdict says the composition is not judged, and when the -// assigner starts and writes it, the checker takes it in then (Checker.Composed). -var ErrAssignerNotRunning = errors.New(delivery.AssignerNotRunning) - -// PresenceEvent is a change of the watchman's presence bucket, or Settled once what stands was handed. -type PresenceEvent struct { - Key string - Online bool - Settled bool -} - -// CompositionEvent is a change of the composition's key (its value, nil for a delete), or Settled once what stands was -// handed. -type CompositionEvent struct { - Value []byte - Settled bool -} - -// AwaitComposition answers the assigner's composition once one is written that does not wait, while an instance of the -// assigner is online by the watchman's presence (`part.mesh-delivery-assigner.`). Once both watches handed -// what stands, no instance online is ErrAssignerNotRunning at once, and so is the last instance going offline during the -// wait. No timer: every answer follows an event; ctx ending is ctx's error. -func AwaitComposition(ctx context.Context, presence <-chan PresenceEvent, comps <-chan CompositionEvent) (delivery.CompositionVerdict, error) { - online := map[string]bool{} - presenceSettled, compSettled := false, false - for { - select { - case <-ctx.Done(): - return delivery.CompositionVerdict{}, ctx.Err() - case e, ok := <-presence: - if !ok { - return delivery.CompositionVerdict{}, errors.New("the watch of the assigner's presence ended") - } - if e.Settled { - presenceSettled = true - break - } - if _, part, instance, err := delivery.ParsePresenceKey(e.Key); err == nil && part == delivery.AssignerModule { - if e.Online { - online[instance] = true - } else { - delete(online, instance) - } - } - case e, ok := <-comps: - if !ok { - return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") - } - if e.Settled { - compSettled = true - break - } - if e.Value == nil { - break - } - v, err := delivery.DecodeCompositionVerdict(e.Value) - if err != nil { - return delivery.CompositionVerdict{}, fmt.Errorf("the composition cannot be read: %w", err) - } - if v.State != delivery.CompositionWaits { - return v, nil - } - } - if presenceSettled && compSettled && len(online) == 0 { - return delivery.CompositionVerdict{}, ErrAssignerNotRunning - } - } -} - -// withComposition is v with the assigner's composition taken in: a failed one fails the check, naming its machine. -func withComposition(v *delivery.Verdict, comp delivery.CompositionVerdict) { - v.Composition = comp.Composition() - v.Took.Composition = comp.Took - if comp.State == delivery.CompositionFail { - machine := comp.Machine - if machine == "" { - machine = "the composition" - } - f := delivery.FailedFirst{Machine: machine, Words: comp.Words} - if v.State == delivery.VerdictFail { - addFailure(v, f) // the tests failed too: the composition is named after them (the operator, 2026-11-02) - return - } - v.State, v.Failed = delivery.VerdictFail, f - } -} - -// notJudged says whether a verdict's composition was not judged because the assigner was not running. -func notJudged(v delivery.Verdict) bool { - return v.Composition != nil && v.Composition.State == delivery.CompositionNotRun && v.Composition.Words == delivery.AssignerNotRunning -} - -// unanswered says whether a verdict failed only on its composition, and that only because nodes the change places did -// not answer its placement in time (delivery.CompositionUnanswered, ADR 0559 §5): its tests passed (the composition is -// its first failure, and a composition is named after every other failure), so a later composition of the same heads is -// taken in on its event, with no new check (ADR 0548 §1: a part of the mesh not heard is judged again when it is heard). -func unanswered(v delivery.Verdict) bool { - return v.State == delivery.VerdictFail && v.Composition != nil && v.Composition.State == delivery.CompositionFail && - strings.Contains(v.Composition.Words, delivery.CompositionUnanswered) && v.Failed.Words == v.Composition.Words -} - -// Composed takes in a composition the assigner wrote after the check's verdict said it was not judged because the -// assigner was not running (the assigner started since), or after it failed only because placed nodes had not answered -// (unanswered): the verdict and its status are written again with it. A composition of heads no candidate holds now, or -// of a verdict that judged its composition otherwise, is left. -func (ch *Checker) Composed(ctx context.Context, comp delivery.CompositionVerdict) { - ch.init() - if comp.State == delivery.CompositionWaits { - return - } - ckey, err := delivery.CandidatesKey(comp.Group) - if err != nil { - return - } - ch.mu.Lock() - c, ok := ch.known[ckey] - ch.mu.Unlock() - if !ok { - return - } - cs := CaseOf(c, ch.heads()) - if cs == nil || cs.HeadsHash != comp.Heads { - return - } - cur, rev, found, err := ch.Verdicts.Get(ctx, cs.Key) - if err != nil || !found { - return - } - v, err := delivery.DecodeVerdict(cur) - if err != nil { - return - } - switch { - case v.State == delivery.VerdictPass && notJudged(v): - case unanswered(v) && comp.Words != v.Composition.Words: - // The nodes answered since (or another failure stands now): the verdict is its tests' pass with this composition. - v.State, v.Failed, v.Failures = delivery.VerdictPass, delivery.FailedFirst{}, nil - default: - return - } - withComposition(&v, comp) - if v.Composition != nil && v.Composition.State != delivery.CompositionNotRun { - v.Composition.FPSet = cs.FPSet - } - data, err := delivery.EncodeVerdict(v) - if err != nil { - ch.logf("%s: the verdict with its composition cannot be written: %v", cs.Key, err) - return - } - if _, err := ch.Verdicts.Update(ctx, cs.Key, data, rev); err != nil { - ch.logf("%s: the verdict with its composition could not be written: %v", cs.Key, err) - return - } - ch.writeOwed(ctx, cs, v, Outcome{Verdict: v}) - ch.logf("%s: the composition came after the verdict: %s", cs.Key, comp.State) -} diff --git a/modules/mesh-delivery-checker/internal/check/awaiting_test.go b/modules/mesh-delivery-checker/internal/check/awaiting_test.go deleted file mode 100644 index c373d0a..0000000 --- a/modules/mesh-delivery-checker/internal/check/awaiting_test.go +++ /dev/null @@ -1,170 +0,0 @@ -package check - -import ( - "context" - "errors" - "strings" - "testing" - "time" - - "git.novox.be/novox/mesh-sdk/go/delivery" -) - -func assignerOn(instance string, online bool) PresenceEvent { - key, _ := delivery.PresencePartKey(delivery.AssignerModule, instance) - return PresenceEvent{Key: key, Online: online} -} - -func compEvent(t *testing.T, v delivery.CompositionVerdict) CompositionEvent { - t.Helper() - data, err := delivery.EncodeCompositionVerdict(v) - if err != nil { - t.Fatal(err) - } - return CompositionEvent{Value: data} -} - -var passed = delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}} - -// 1297/B3d: with no assigner online, the wait answers ErrAssignerNotRunning as soon as both watches handed what stands. -func TestNoAssignerAnswersAtOnce(t *testing.T) { - presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) - presence <- assignerOn("novox", false) - presence <- PresenceEvent{Settled: true} - comps <- CompositionEvent{Settled: true} - ctx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - if _, err := AwaitComposition(ctx, presence, comps); !errors.Is(err, ErrAssignerNotRunning) { - t.Fatalf("answered %v, want the assigner not running", err) - } -} - -// 1297/B3d: with the assigner online it waits for the key, past a composition that waits, and answers the verdict. -func TestAnOnlineAssignerIsWaitedFor(t *testing.T) { - presence, comps := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) - presence <- assignerOn("novox", true) - presence <- PresenceEvent{Settled: true} - comps <- CompositionEvent{Settled: true} - done := make(chan error, 1) - var got delivery.CompositionVerdict - go func() { - var err error - got, err = AwaitComposition(context.Background(), presence, comps) - done <- err - }() - comps <- compEvent(t, delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "w"}) - select { - case err := <-done: - t.Fatalf("answered before the composition was written: %v", err) - case <-time.After(50 * time.Millisecond): - } - comps <- compEvent(t, passed) - if err := <-done; err != nil || got.State != delivery.CompositionPass { - t.Fatalf("answered %+v, %v", got, err) - } - // And the last instance going offline during a wait ends it as not running. - presence2, comps2 := make(chan PresenceEvent, 4), make(chan CompositionEvent, 4) - presence2 <- assignerOn("novox", true) - presence2 <- PresenceEvent{Settled: true} - comps2 <- CompositionEvent{Settled: true} - presence2 <- assignerOn("novox", false) - if _, err := AwaitComposition(context.Background(), presence2, comps2); !errors.Is(err, ErrAssignerNotRunning) { - t.Errorf("the assigner going offline answered %v", err) - } -} - -// 1297/B3d: a passing check with no assigner says at once, loud, that its composition is not judged; never a pass of it. -func TestAPassingCheckWithNoAssignerSaysTheCompositionIsNotJudged(t *testing.T) { - b := newBench(t) - b.ex.Compositions = &fakeCompositions{err: ErrAssignerNotRunning} - v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict - if v.Composition == nil || v.Composition.State != delivery.CompositionNotRun || v.Composition.Words != delivery.AssignerNotRunning { - t.Fatalf("composition = %+v", v.Composition) - } - if words := compositionWords(v.Composition, 0); words != "composition not judged: the assigner is not running" { - t.Errorf("the line says %q", words) - } -} - -// 1297/B3d: the assigner starting writes the composition later, and the checker takes it into the verdict that said it -// was not judged, writing the status again; a failed one fails the check. -func TestTheAssignerStartingBringsTheComposition(t *testing.T) { - for _, comp := range []delivery.CompositionVerdict{passed, - {Group: "feat/x", Heads: "", State: delivery.CompositionFail, SDK: "x", Machine: "g14", Words: "g14 runs web, which the change removes"}} { - ch, cs := checkerWithNotJudged(t) - comp.Heads = cs.HeadsHash - ch.Composed(context.Background(), comp) - raw, _, _, _ := ch.Verdicts.Get(context.Background(), cs.Key) - v, err := delivery.DecodeVerdict(raw) - if err != nil || v.Composition == nil || v.Composition.State != comp.State { - t.Fatalf("after the composition came, the verdict is %+v, %v", v, err) - } - if comp.State == delivery.CompositionFail && (v.State != delivery.VerdictFail || v.Failed.Machine != "g14") { - t.Errorf("a failed composition left the verdict %+v", v) - } - owed := ch.Owed.(*memKV) - keys, _ := owed.Keys(context.Background()) - if len(keys) == 0 { - t.Fatal("no status was written again") - } - data, _, _, _ := owed.Get(context.Background(), keys[0]) - o, _ := delivery.DecodeCheckerOwed(data) - if !strings.Contains(o.Text, "composition passed") && !strings.Contains(o.Text, "g14 runs web") { - t.Errorf("the status's line says %q", o.Text) - } - } - - // ADR 0559 §5: a verdict that failed only because a placed node had not answered takes the same heads' later - // composition in on its event, a pass making it pass with no new check; a composition failing the same way is left. - ch, cs := checkerWithNotJudged(t) - silent := delivery.CompositionVerdict{Group: cs.Group, Heads: cs.HeadsHash, State: delivery.CompositionFail, SDK: "x", Machine: "g14", - Words: "g14 (offline) " + delivery.CompositionUnanswered} - ch.Composed(context.Background(), silent) - for _, row := range []struct { - name string - comp delivery.CompositionVerdict - state string - }{ - {"the same failure again", silent, delivery.VerdictFail}, - {"the node's answer", delivery.CompositionVerdict{Group: cs.Group, Heads: cs.HeadsHash, State: delivery.CompositionPass, SDK: "x"}, delivery.VerdictPass}, - } { - ch.Composed(context.Background(), row.comp) - raw, _, _, _ := ch.Verdicts.Get(context.Background(), cs.Key) - if v, err := delivery.DecodeVerdict(raw); err != nil || v.State != row.state || v.Composition == nil || v.Composition.State != row.comp.State { - t.Errorf("%s: the verdict is %+v, %v", row.name, v, err) - } - } -} - -// checkerWithNotJudged is a checker holding a candidate whose passing verdict said its composition was not judged. -func checkerWithNotJudged(t *testing.T) (*Checker, *Case) { - t.Helper() - cs := caseOf(t, pullOf(repoA, 7, head1)) - raw := candidateOf(t, cs.Group, pullOf(repoA, 7, head1)) - ch := &Checker{Verdicts: newMemKV(), Owed: newMemKV(), Trunks: fixedTrunks{cs.Trunk}} - ckey, _ := delivery.CandidatesKey(cs.Group) - ch.init() - c, err := delivery.DecodeCandidate(raw) - if err != nil { - t.Fatal(err) - } - ch.known[ckey] = c - if got := CaseOf(c, ch.heads()); got == nil || got.Key != cs.Key { - t.Fatalf("the case is %+v, want %s", got, cs.Key) - } - v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, State: delivery.VerdictPass, - Levels: map[string]delivery.LevelResult{delivery.LevelUnit: {Run: 1}, delivery.LevelIntegration: {Run: 1}}, - Composition: &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning}} - data, err := delivery.EncodeVerdict(v) - if err != nil { - t.Fatal(err) - } - if _, err := ch.Verdicts.Create(context.Background(), cs.Key, data); err != nil { - t.Fatal(err) - } - return ch, cs -} - -type fixedTrunks struct{ heads map[string]string } - -func (f fixedTrunks) Heads() map[string]string { return f.heads } diff --git a/modules/mesh-delivery-checker/internal/check/composition.go b/modules/mesh-delivery-checker/internal/check/composition.go new file mode 100644 index 0000000..0501844 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/composition.go @@ -0,0 +1,28 @@ +package check + +import ( + "strings" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// withComposition is v with the check's composition taken in: a failed one fails the check, naming its machine. +func withComposition(v *delivery.Verdict, comp delivery.CompositionVerdict) { + v.Composition = comp.Composition() + v.Took.Composition = comp.Took + if comp.State == delivery.CompositionPass && len(comp.Warnings) > 0 && v.Composition != nil { + v.Composition.Words = strings.Join(comp.Warnings, "; ") // passed with a warning (ADR 0593 §9) + } + if comp.State == delivery.CompositionFail { + machine := comp.Machine + if machine == "" { + machine = "the composition" + } + f := delivery.FailedFirst{Machine: machine, Words: comp.Words} + if v.State == delivery.VerdictFail { + addFailure(v, f) // the tests failed too: the composition is named after them (the operator, 2026-11-02) + return + } + v.State, v.Failed = delivery.VerdictFail, f + } +} diff --git a/modules/mesh-delivery-checker/internal/check/composition_test.go b/modules/mesh-delivery-checker/internal/check/composition_test.go index bd5d357..5e3e6d4 100644 --- a/modules/mesh-delivery-checker/internal/check/composition_test.go +++ b/modules/mesh-delivery-checker/internal/check/composition_test.go @@ -9,18 +9,17 @@ import ( "git.novox.be/novox/mesh-sdk/go/delivery" ) -// fakeCompositions answers the assigner's composition of a check from memory, recording the keys asked. +// fakeCompositions answers a check's composition from memory, recording the keys asked. type fakeCompositions struct { v delivery.CompositionVerdict err error asked []string - // waits answers only that the composition waits, as the assigner does for a member the indexer holds no manifest of: - // Await returns only when its context ends. + // waits answers only when its context ends. waits bool } -func (f *fakeCompositions) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { - f.asked = append(f.asked, key) +func (f *fakeCompositions) Compose(ctx context.Context, c *Case) (delivery.CompositionVerdict, error) { + f.asked = append(f.asked, c.Key) if f.waits { <-ctx.Done() return delivery.CompositionVerdict{}, ctx.Err() @@ -95,3 +94,8 @@ func TestACompositionNotHadIsNotRun(t *testing.T) { t.Errorf("a try asked for a composition: %v", f2.asked) } } + +// fixedTrunks are trunk heads that never move. +type fixedTrunks struct{ heads map[string]string } + +func (f fixedTrunks) Heads() map[string]string { return f.heads } diff --git a/modules/mesh-delivery-checker/internal/check/engines.go b/modules/mesh-delivery-checker/internal/check/engines.go new file mode 100644 index 0000000..ae38d8e --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/engines.go @@ -0,0 +1,228 @@ +package check + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/composer" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **A check's composition is the checker's, made from what each node's engine answers** (novox/hq ADR 0593 §7, §9). +// Only a node knows whether a change composes there: the checker asks every placed node's engine +// `node-engine.compose-candidate` in parallel, with the check's candidate, and waits at most its answer bound +// (compose-answer-bound, 60 s by default). A node that refuses fails the check, naming the node and each module with the +// engine's words; a node that does not answer within the bound, answers unready or cannot be asked is named, "not composed +// on that node: no answer", and the check passes with that warning: its own composition judges again when it applies, +// and the canary rule holds. The checker decides by the answer's verdict, never by its words. +// +// **A group that changes the node-engine's composition code** (a member of the node-engine's repository) is not judged by +// the running engines, which cannot judge their own next code: each node's engine answers the composition input it +// composed from (ComposeCandidateAsk.Input), and the composer program built from that member's head composes it in a +// build module's sandbox (Sandbox), with every manifest of the catalogue as the case leaves it and the indexer's verdicts. + +// EngineAsker asks one node's engine to compose a check's candidate (the verb node-engine.compose-candidate on that node). +type EngineAsker interface { + ComposeCandidate(ctx context.Context, node string, ask delivery.ComposeCandidateAsk) (delivery.ComposeCandidateAnswer, error) +} + +// Sandbox runs a compose job on the Go build seat and answers the composer's verdict. +type Sandbox interface { + Compose(ctx context.Context, job delivery.ComposeJob) (composer.Verdict, error) +} + +// EngineCompositions is the checker's composition of a check (Examiner.Compositions). +type EngineCompositions struct { + Asker EngineAsker + // Placed names every node in placement now (the seat mesh-placement's state). + Placed func() []string + // Bound is the answer bound (the setting compose-answer-bound); zero is delivery.ComposeAnswerBound. + Bound time.Duration + // Composer is the commit of the node-engine the mesh runs, named as what composed (ADR 0548 §6's running commit). + Composer func() string + // Sandbox runs the composer of a node-engine member's head; nil leaves such a group judged by the running engines, + // said in a warning. + Sandbox Sandbox + // Manifests are the indexer's trunk manifest digests by module, and Verdicts its refusals of them: the catalogue the + // sandbox composes with, before the case's own manifests are laid over. + Manifests TrunkManifests + Verdicts func() map[string]delivery.ManifestVerdict +} + +// composerHead is the node-engine member's head of a case, or empty. +func composerHead(c *Case) string { + if m := c.member(nodeEngineRepo); m != nil { + return m.Head + } + return "" +} + +// Compose asks every placed node and judges their answers (JudgeEngines). An error is only a case it cannot ask at all. +func (e EngineCompositions) Compose(ctx context.Context, c *Case) (delivery.CompositionVerdict, error) { + began := time.Now() + key, err := delivery.CompositionKey(c.Group, c.HeadsHash) + if err != nil { + return delivery.CompositionVerdict{}, err + } + var nodes []string + if e.Placed != nil { + nodes = e.Placed() + } + bound := e.Bound + if bound <= 0 { + bound = delivery.ComposeAnswerBound + } + head := composerHead(c) + sandboxed := head != "" && e.Sandbox != nil + answers := map[string]delivery.ComposeCandidateAnswer{} + silent := map[string]string{} + var mu sync.Mutex + var wg sync.WaitGroup + for _, node := range nodes { + wg.Add(1) + go func() { + defer wg.Done() + at, cancel := context.WithTimeout(ctx, bound) + defer cancel() + a, err := e.Asker.ComposeCandidate(at, node, delivery.ComposeCandidateAsk{Candidate: key, Input: sandboxed}) + if err == nil && sandboxed && a.Verdict != delivery.ComposeCandidateUnready { + // The composer's build and run are the check's to bound, not the engine's answer bound. + a, err = e.inSandbox(ctx, c, head, node, a) + } + mu.Lock() + defer mu.Unlock() + switch { + case err != nil: + silent[node] = err.Error() + case a.Verdict == delivery.ComposeCandidateUnready: + silent[node] = "unready" + default: + answers[node] = a + } + }() + } + wg.Wait() + // What composed, named by the node-engine's commit: the member's head in the sandbox, else the one the mesh runs. + composerName := "node-engine" + if sandboxed { + composerName = head + } else if e.Composer != nil && e.Composer() != "" { + composerName = e.Composer() + } + v := JudgeEngines(c.Group, c.HeadsHash, composerName, nodes, answers, silent) + if head != "" && e.Sandbox == nil { + v.Warnings = append(v.Warnings, "the group changes the node-engine's composition code and was composed by the running engines") + } + v.Took = time.Since(began).Milliseconds() + return v, nil +} + +// JudgeEngines is the check's composition verdict from the engines' answers (ADR 0593 §7, §9): failed when any node's +// answer is refused, naming each node and each module refused with its words (the first refusing node is the machine); +// passed otherwise. Every node in nodes without an answer (silent, by node with why, for the journal only) is named +// with delivery.ComposeNoAnswer as a warning. composer names what composed, for the line. Pure. +func JudgeEngines(group, heads, composerName string, nodes []string, answers map[string]delivery.ComposeCandidateAnswer, silent map[string]string) delivery.CompositionVerdict { + v := delivery.CompositionVerdict{Group: group, Heads: heads, State: delivery.CompositionPass, SDK: composerName} + sorted := append([]string(nil), nodes...) + sort.Strings(sorted) + var fails []string + for _, node := range sorted { + a, answered := answers[node] + if !answered { + v.Warnings = append(v.Warnings, node+": "+delivery.ComposeNoAnswer) + continue + } + v.Machines = append(v.Machines, node) + if a.Verdict != delivery.ComposeCandidateRefused { + continue + } + if v.Machine == "" { + v.Machine = node + } + var words []string + if a.Refused != "" { + words = append(words, a.Refused) + } + for _, m := range append(append([]delivery.ModuleComposed(nil), a.Modules...), a.Own) { + if m.State == delivery.ModuleRefusedState { + words = append(words, m.Module+": "+m.Words) + } + } + if len(words) == 0 { + words = append(words, "refused, naming no module") + } + fails = append(fails, "not composed on "+node+": "+strings.Join(words, "; ")) + } + if len(fails) > 0 { + v.State, v.Words = delivery.CompositionFail, strings.Join(fails, "; ") + } + if len(sorted) == 0 { + v.Warnings = append(v.Warnings, "no node is placed, so the change was composed on none") + } + return v +} + +// inSandbox composes a node's answer again with the composer built from the node-engine member's head, over the input the +// node's engine composed from: the node's answer becomes the sandbox's, refused with the composer's words where the +// composer refuses the node. +func (e EngineCompositions) inSandbox(ctx context.Context, c *Case, head, node string, a delivery.ComposeCandidateAnswer) (delivery.ComposeCandidateAnswer, error) { + if len(a.Input) == 0 { + return a, fmt.Errorf("the engine of %s answered no composition input, so the node-engine's head cannot compose it", node) + } + manifests := map[string]composer.Manifest{} + if e.Manifests != nil { + for name, digest := range e.Manifests.ManifestDigests() { + manifests[name] = composer.Manifest{Digest: digest} + } + } + for _, p := range c.Pulls { + for name, m := range p.Fps { + if m.Manifest != "" { + manifests[name] = composer.Manifest{Digest: m.Manifest} + } + } + } + verdicts := map[string]delivery.ManifestVerdict{} + if e.Verdicts != nil { + for name, v := range e.Verdicts() { + verdicts[name] = v + } + } + for _, p := range c.Pulls { + for name := range p.Fps { + delete(verdicts, name) // a manifest the change brings is the checker's own judge's (ADR 0578 §2) + } + } + req := composer.Request{V: composer.Version, Input: a.Input, Depth: composer.DepthCompose, Machines: []string{node}, + Cases: []composer.Case{{Name: "change", Manifests: manifests, Verdicts: verdicts}}} + raw, err := json.Marshal(req) + if err != nil { + return a, err + } + job := delivery.ComposeJob{Fetch: map[string]any{"repository": nodeEngineRepo, "commit": head}, Dir: delivery.RootPath, + Package: composer.Package, Request: raw, Reader: "the node-engine head of " + c.Group} + verdict, err := e.Sandbox.Compose(ctx, job) + if err != nil { + return a, fmt.Errorf("the composer at the node-engine's head %s did not answer for %s: %w", head, node, err) + } + out := delivery.ComposeCandidateAnswer{V: a.V, Node: node, Candidate: a.Candidate, Verdict: delivery.ComposeCandidateComposed} + for _, cv := range verdict.Cases { + if m, ok := cv.Machines[node]; ok && !m.Composes { + out.Verdict = delivery.ComposeCandidateRefused + out.Refused = strings.Join(append(append([]string{m.Reason}, m.Problems...), m.Refusals...), "; ") + } + for name, read := range cv.Manifests { + if !read.Accepted { + out.Verdict = delivery.ComposeCandidateRefused + out.Modules = append(out.Modules, delivery.ModuleComposed{Module: name, State: delivery.ModuleRefusedState, Words: read.Refused}) + } + } + } + sort.Slice(out.Modules, func(i, j int) bool { return out.Modules[i].Module < out.Modules[j].Module }) + return out, nil +} diff --git a/modules/mesh-delivery-checker/internal/check/engines_test.go b/modules/mesh-delivery-checker/internal/check/engines_test.go new file mode 100644 index 0000000..76f7149 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/engines_test.go @@ -0,0 +1,168 @@ +package check + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/composer" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeEngines answers compose-candidate by node from memory; a node it holds no answer for never answers. +type fakeEngines struct { + mu sync.Mutex + answers map[string]delivery.ComposeCandidateAnswer + errs map[string]error + asked map[string]delivery.ComposeCandidateAsk +} + +func (f *fakeEngines) ComposeCandidate(ctx context.Context, node string, ask delivery.ComposeCandidateAsk) (delivery.ComposeCandidateAnswer, error) { + f.mu.Lock() + if f.asked == nil { + f.asked = map[string]delivery.ComposeCandidateAsk{} + } + f.asked[node] = ask + a, ok := f.answers[node] + err := f.errs[node] + f.mu.Unlock() + if err != nil { + return delivery.ComposeCandidateAnswer{}, err + } + if !ok { + <-ctx.Done() + return delivery.ComposeCandidateAnswer{}, ctx.Err() + } + return a, nil +} + +// fakeSandbox answers a compose job with a verdict, recording the job. +type fakeSandbox struct { + mu sync.Mutex + jobs []delivery.ComposeJob + v func(node string) composer.Verdict +} + +func (f *fakeSandbox) Compose(_ context.Context, job delivery.ComposeJob) (composer.Verdict, error) { + var req composer.Request + if err := json.Unmarshal(job.Request, &req); err != nil { + return composer.Verdict{}, err + } + f.mu.Lock() + f.jobs = append(f.jobs, job) + f.mu.Unlock() + return f.v(req.Machines[0]), nil +} + +type digests map[string]string + +func (d digests) ManifestDigests() map[string]string { return d } + +// ADR 0593 §7, §9 (D12, G1, G2): the checker asks every placed node's engine in parallel and makes the composition from +// their answers: a refusal fails it naming the node and each module; a node that does not answer within the bound, +// answers unready or cannot be asked is named "not composed on that node: no answer" and the check passes with that +// warning, decided by the verdict and never by its words; a group with a node-engine member is composed by the composer +// built at that member's head, over the input each node's engine composed from. +func TestTheCompositionIsMadeFromTheEnginesAnswers(t *testing.T) { + composed := func(node string) delivery.ComposeCandidateAnswer { + return delivery.ComposeCandidateAnswer{Node: node, Verdict: delivery.ComposeCandidateComposed, + Modules: []delivery.ModuleComposed{{Module: "web", State: delivery.ModuleComposedState, Digest: "sha256:" + strings.Repeat("a", 64)}}, + Own: delivery.ModuleComposed{Module: "node", State: delivery.ModuleComposedState, Digest: "sha256:" + strings.Repeat("b", 64)}} + } + cs := caseOf(t, pullOf(repoA, 7, head1)) + key, _ := delivery.CompositionKey(cs.Group, cs.HeadsHash) + compose := func(e *fakeEngines, nodes ...string) delivery.CompositionVerdict { + t.Helper() + ec := EngineCompositions{Asker: e, Placed: func() []string { return nodes }, Bound: 50 * time.Millisecond, + Composer: func() string { return strings.Repeat("7", 40) }} + v, err := ec.Compose(context.Background(), cs) + if err != nil { + t.Fatal(err) + } + return v + } + + t.Run("one node refusing fails the composition, naming the node and the module", func(t *testing.T) { + refusing := composed("laptop") + refusing.Verdict = delivery.ComposeCandidateRefused + refusing.Modules = []delivery.ModuleComposed{{Module: "web", State: delivery.ModuleRefusedState, Words: "a required value is unset"}, + {Module: "mail", State: delivery.ModuleHeldState, Words: "held as today"}} + e := &fakeEngines{answers: map[string]delivery.ComposeCandidateAnswer{"anchor": composed("anchor"), "laptop": refusing}} + v := compose(e, "anchor", "laptop") + if v.State != delivery.CompositionFail || v.Machine != "laptop" || !strings.Contains(v.Words, "not composed on laptop: web: a required value is unset") || + strings.Contains(v.Words, "mail") || len(v.Machines) != 2 { + t.Errorf("verdict = %+v", v) + } + if e.asked["anchor"].Candidate != key || e.asked["laptop"].Input { + t.Errorf("asked %+v, want the check's candidate %s and no input", e.asked, key) + } + var ver delivery.Verdict + withComposition(&ver, v) + if ver.State != delivery.VerdictFail || ver.Failed.Machine != "laptop" { + t.Errorf("the check's verdict = %+v", ver) + } + }) + + t.Run("a silent, unready or unreachable node is no answer, and the check passes naming it", func(t *testing.T) { + unready := delivery.ComposeCandidateAnswer{Node: "g14", Verdict: delivery.ComposeCandidateUnready, Refused: "a module refused it"} + e := &fakeEngines{answers: map[string]delivery.ComposeCandidateAnswer{"anchor": composed("anchor"), "g14": unready}, + errs: map[string]error{"ace": errors.New("no responders")}} + began := time.Now() + v := compose(e, "anchor", "laptop", "g14", "ace") + if took := time.Since(began); took > 2*time.Second { + t.Errorf("the nodes were not asked in parallel within the bound: %s", took) + } + if v.State != delivery.CompositionPass || v.SDK != strings.Repeat("7", 40) || len(v.Machines) != 1 { + t.Fatalf("verdict = %+v", v) + } + for _, node := range []string{"ace", "g14", "laptop"} { + if !strings.Contains(strings.Join(v.Warnings, "; "), node+": "+delivery.ComposeNoAnswer) { + t.Errorf("%s is not named: %v", node, v.Warnings) + } + } + var ver delivery.Verdict + ver.State = delivery.VerdictPass + withComposition(&ver, v) + if ver.State != delivery.VerdictPass || !strings.Contains(composition(ver.Composition, 0, false), "laptop: "+delivery.ComposeNoAnswer) { + t.Errorf("the line = %q", composition(ver.Composition, 0, false)) + } + }) + + t.Run("a node-engine member is composed by the composer at its head over each node's input", func(t *testing.T) { + c, err := delivery.DecodeCandidate(candidateOf(t, "feat/x", pullOf(nodeEngineRepo, 3, head2), pullOf(repoA, 7, head1))) + if err != nil { + t.Fatal(err) + } + host := CaseOf(c, map[string]string{repoA: trunkA, nodeEngineRepo: trunkA}) + withInput := composed("anchor") + withInput.Input = json.RawMessage(`{"nodes":[{"name":"anchor"}]}`) + e := &fakeEngines{answers: map[string]delivery.ComposeCandidateAnswer{"anchor": withInput}} + sb := &fakeSandbox{v: func(node string) composer.Verdict { + return composer.Verdict{V: composer.Version, Cases: []composer.CaseVerdict{{Name: "change", + Machines: map[string]composer.Machine{node: {Composes: false, Reason: "the new composer refuses it"}}}}} + }} + ec := EngineCompositions{Asker: e, Placed: func() []string { return []string{"anchor"} }, Bound: time.Second, + Sandbox: sb, Manifests: digests{"web": "sha256:" + strings.Repeat("c", 64)}} + v, err := ec.Compose(context.Background(), host) + if err != nil { + t.Fatal(err) + } + if !e.asked["anchor"].Input || len(sb.jobs) != 1 { + t.Fatalf("asked %+v, jobs %+v: want the input asked and one job", e.asked, sb.jobs) + } + job := sb.jobs[0] + var req composer.Request + _ = json.Unmarshal(job.Request, &req) + if job.Fetch["repository"] != nodeEngineRepo || job.Fetch["commit"] != head2 || job.Package != composer.Package || + string(req.Input) != string(withInput.Input) || req.Cases[0].Manifests["web"].Digest == "" { + t.Errorf("job = %+v, request %+v", job, req) + } + if v.State != delivery.CompositionFail || v.SDK != head2 || !strings.Contains(v.Words, "the new composer refuses it") { + t.Errorf("verdict = %+v", v) + } + }) +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 5547d47..5328576 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -3,7 +3,6 @@ package check import ( "context" "encoding/json" - "errors" "fmt" "path" "slices" @@ -117,9 +116,8 @@ type Examiner struct { Manifests TrunkManifests // Predefined reads the catalogue's predefined keys the indexer keeps (manifests_judge.go); nil reads none. Predefined PredefinedList - // 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 makes a check's composition (ADR 0593 §7): the engines' answers (EngineCompositions); nil leaves the + // composition not run. Compositions CompositionReader // Nodes names every node with a record, read from each `.node` of the node-engine's have (novox/hq ADR 0567 // §12): every check ask carries them (CheckJob.NodeNames) for the names check, so no repository lists them. A check @@ -131,10 +129,9 @@ type Examiner struct { 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. +// CompositionReader makes a check's composition verdict; an error is only a case that cannot be composed at all. type CompositionReader interface { - Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) + Compose(ctx context.Context, c *Case) (delivery.CompositionVerdict, error) } func (j *Examiner) now() time.Time { @@ -260,9 +257,9 @@ func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { 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; with no assigner running it is not judged, said at once. +// composed makes the check's composition from the engines' answers into its verdict (ADR 0593 §7, §9): a failed +// composition fails the check, naming the machine it failed at; a node that did not answer is a warning. A try is not +// composed (it has no candidate); a composition that cannot be made leaves it not run, said. func (st *state) composed(ctx context.Context, v *delivery.Verdict) { if st.c.Composition != nil && st.c.Try == nil { // A carry whose candidate's fingerprints did not change keeps the composition it passed (ADR 0548 §5). @@ -275,18 +272,14 @@ func (st *state) composed(ctx context.Context, v *delivery.Verdict) { return } if len(st.unpreviewed) > 0 { - // The assigner would only say it waits for a manifest the indexer will not have: the check does not wait for it. + // An engine would only say it cannot read a manifest the indexer will not have: the check does not ask it. v.Composition = &delivery.Composition{State: delivery.CompositionNotRun, Words: "no preview of a merge of " + strings.Join(st.unpreviewed, ", ") + " (its head does not merge or its fetch did not answer), so the change cannot be composed"} return } - comp, err := st.j.Compositions.Await(ctx, st.c.Key) - switch { - case errors.Is(err, ErrAssignerNotRunning): - v.Composition = &delivery.Composition{State: delivery.CompositionNotRun, Words: delivery.AssignerNotRunning} - return - case err != nil: - st.j.logf("%s: the composition could not be read, so it is said as not run: %v", st.c.Key, err) + comp, err := st.j.Compositions.Compose(ctx, st.c) + if err != nil { + st.j.logf("%s: the composition could not be made, so it is said as not run: %v", st.c.Key, err) return } withComposition(v, comp) @@ -315,7 +308,7 @@ func (st *state) outcome(v delivery.Verdict) Outcome { v.OverBound = append(v.OverBound, level) } } - // The composition is the assigner's check (issue 1297): until it delivers a verdict says it was not run (ADR 0484 clause 6). + // The composition is made from the engines' answers (ADR 0593 §7): until it is, a verdict says it was not run (ADR 0484 clause 6). v.Composition = &delivery.Composition{State: delivery.CompositionNotRun} var slow []string for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} { diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go index 31e9f14..153b5b6 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -156,8 +156,8 @@ func levelWords(name string, r delivery.LevelResult) string { return s } -// compositionWords says the composition: not yet run, or once the assigner's check reports it (issue 1297) passed or -// failed, with the time it took and the machines it composed (issue 1300). +// compositionWords says the composition: not yet run, or once the engines answered (ADR 0593 §7) passed, with the nodes +// that gave no answer, or failed, with the time it took and the machines it composed (issue 1300). func compositionWords(c *delivery.Composition, took int64) string { return composition(c, took, false) } // composition is compositionWords; short counts the machines and leaves the SDK commit to the report. @@ -183,11 +183,15 @@ func composition(c *delivery.Composition, took int64, short bool) string { on = fmt.Sprintf(" on %d machines: %s", len(machines), strings.Join(machines, ", ")) } } + warned := "" + if c.State == delivery.CompositionPass && c.Words != "" { + warned = " (" + c.Words + ")" // a node that did not answer (ADR 0593 §9) + } switch { case c.State == delivery.CompositionPass && short: - return "composition passed" + in + on // the SDK it composed with is in the report + return "composition passed" + in + on + warned // what it composed with is in the report case c.State == delivery.CompositionPass: - return "composition passed" + in + " with " + short12(c.SDK) + on + return "composition passed" + in + " with " + short12(c.SDK) + on + warned } return "composition failed" + in + on + ": " + c.Words } diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 82cd884..df61846 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -54,10 +54,13 @@ type manifest struct { Uses []string `json:"uses"` Invokes []string `json:"invokes"` Config map[string]struct { - Type string `json:"type"` - Description string `json:"description"` - Pattern string `json:"pattern"` - Default string `json:"default"` + Type string `json:"type"` + Description string `json:"description"` + Pattern string `json:"pattern"` + Default any `json:"default"` + Minimum *float64 `json:"minimum"` + Maximum *float64 `json:"maximum"` + Trusted bool `json:"trusted"` } `json:"config"` OwnSecrets map[string]string `json:"own-secrets"` SecretsOwner string `json:"secrets-owner"` @@ -102,14 +105,17 @@ 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-delivery-assigner.compositions", "mesh-delivery-assigner.placement", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-indexer.candidates", + "mesh-delivery-assigner.placement", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-gate.turns", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.packages", "mesh-delivery-indexer.predefined", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } // 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no // build module's tool and holds no capacity: it puts check asks on the Go and Node build seats (ADR 0525 §1, §4). - if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", "gitea.gitea_list_branches"}; !reflect.DeepEqual(sorted(m.Invokes), want) { - t.Errorf("invokes = %v, want only the forge's reads for a mesh-sdk pin (ADR 0484 clause 9)", m.Invokes) + // The forge's reads for a mesh-sdk pin (ADR 0484 clause 9), and of the node seat node-engine its one verb + // compose-candidate, asked of every placed node (ADR 0593 §7), never the seat's other verbs. + if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", "gitea.gitea_list_branches", + "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) { + t.Errorf("invokes = %v, want the forge's reads and compose-candidate", m.Invokes) } if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) { t.Errorf("uses = %v, want the Go and Node build seats", m.Uses) @@ -139,7 +145,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 !reflect.DeepEqual(m.Claims[0].Serves, []string{"try", "try-read", "try-end"}) { t.Errorf("claims = %+v, want the mesh seat mesh-delivery-checker serving try, try-read and try-end", m.Claims) } - for _, k := range []string{"logs", "emits", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} { + for _, k := range []string{"logs", "emits", "provides", "requires", "defines-seats", "seat", "seats", "bus"} { // config: compose-answer-bound (ADR 0593 §7) if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the checker has none", k) } @@ -165,14 +171,19 @@ func TestManifestProcess(t *testing.T) { // 1678/AC1 t.Errorf("process = %v", p) } env, _ := p["env"].(map[string]any) - if len(env) != 1 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" { - t.Errorf("env = %v, want the broker file alone", p["env"]) + if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_CHECKER_SETTINGS"] != "${dir:state}/settings.json" { + t.Errorf("env = %v, want the broker file and the settings file", p["env"]) } + // Its settings hold its answer bound alone (ADR 0593 §7: compose-answer-bound, 60 s, the checker's), never a capacity + // of its own (ADR 0525 §4). for _, r := range m.Resources { - if r["id"] == "settings" { - t.Errorf("a settings resource %v: the checker has no capacity of its own (ADR 0525 §4)", r) + if r["id"] == "settings" && !reflect.DeepEqual(r["takes-config"], []any{"compose-answer-bound"}) { + t.Errorf("the settings resource %v takes more than compose-answer-bound", r) } } + if c, ok := m.Config["compose-answer-bound"]; !ok || c.Default != float64(60) || c.Trusted { + t.Errorf("compose-answer-bound = %+v, want 60 by default, not trusted", c) + } if _, bus := p["bus"]; bus { t.Errorf("the process declares a bus: it runs on its own broker credential") } diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 841c990..23247f5 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -34,7 +34,6 @@ "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-delivery-assigner.compositions", "mesh-delivery-assigner.placement", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", @@ -66,13 +65,24 @@ "gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", - "gitea.gitea_list_branches" + "gitea.gitea_list_branches", + "seat:node-engine.compose-candidate" ], "own-secrets": { "broker": "${dir:state}/broker" }, "secrets-owner": "mesh-delivery-checker", + "config": { + "compose-answer-bound": { + "type": "number", + "description": "seconds a check waits for each placed node's engine to answer compose-candidate (novox/hq ADR 0593 §7, §9); a node that has not answered by then is named in the verdict as not composed on that node: no answer, and the check passes with that warning.", + "minimum": 1, + "maximum": 600, + "trusted": false, + "default": 60 + } + }, "resources": [ { "id": "account", @@ -89,6 +99,18 @@ "place": ".", "unknown": "program" }, + { + "id": "settings", + "type": "file", + "path": "${dir:state}/settings.json", + "mode": "0600", + "owner": "mesh-delivery-checker", + "takes-config": [ + "compose-answer-bound" + ], + "merge": "json", + "template": "files/settings.json" + }, { "id": "checker", "type": "process", @@ -98,7 +120,8 @@ "./mesh-delivery-checker" ], "env": { - "MESH_BROKER_FILE": "${dir:state}/broker" + "MESH_BROKER_FILE": "${dir:state}/broker", + "MESH_CHECKER_SETTINGS": "${dir:state}/settings.json" }, "user": "mesh-delivery-checker", "health": {