mesh-delivery-checker: make a check's composition from every placed node's engine answer
ADR 0593 §7, §9: the checker asks compose-candidate of each placed node in parallel within its own setting compose-answer-bound (60 s); a refusal fails the check naming node and module, a silent, unready or unreachable node passes with "not composed on that node: no answer". A group with a node-engine member is composed by the composer at that head on the Go build seat over each node's input. The wait on the assigner's compositions goes, and the checker invokes the one engine verb it needs. Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
-64
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
-107
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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{}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
{
|
||||
"compose-answer-bound": 60
|
||||
}
|
||||
@@ -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.<instance>`). 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)
|
||||
}
|
||||
@@ -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 }
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
@@ -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 }
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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)
|
||||
}
|
||||
})
|
||||
}
|
||||
@@ -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>.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} {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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": {
|
||||
|
||||
Reference in New Issue
Block a user