Merge branch 'feat/r4g-assigner' into feat/r4g
This commit is contained in:
@@ -0,0 +1,276 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"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/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events
|
||||
// are kept in, as the controller's broker raises them.
|
||||
type bed struct {
|
||||
srv *server.Server
|
||||
js jetstream.JetStream
|
||||
ctx context.Context
|
||||
dir string
|
||||
buckets map[string]jetstream.KeyValue
|
||||
}
|
||||
|
||||
func startBed(t *testing.T, leaveOut string) *bed {
|
||||
t.Helper()
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "assigner")
|
||||
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, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}}
|
||||
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||
if name == leaveOut {
|
||||
continue
|
||||
}
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.buckets[name] = kv
|
||||
}
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return b
|
||||
}
|
||||
|
||||
// start runs the program as the node does, through run with its environment and a broker file; it answers the
|
||||
// channel run's error arrives on.
|
||||
func (b *bed) start(t *testing.T) <-chan error {
|
||||
t.Helper()
|
||||
broker := filepath.Join(b.dir, "broker")
|
||||
cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-assigner", User: "n1.mesh-delivery-assigner", Password: "pw"})
|
||||
if err := os.WriteFile(broker, cred, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
env := map[string]string{"MESH_BROKER_FILE": broker}
|
||||
ctx, stop := context.WithCancel(b.ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() { done <- run(ctx, func(k string) string { return env[k] }, io.Discard) }()
|
||||
t.Cleanup(func() {
|
||||
stop()
|
||||
select {
|
||||
case <-done:
|
||||
case <-time.After(10 * time.Second):
|
||||
}
|
||||
})
|
||||
return done
|
||||
}
|
||||
|
||||
// 1297/B2b: started whole on its credential, the assigner says on the bus the three states it keeps, each readable,
|
||||
// having opened every bucket it names.
|
||||
func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
done := b.start(t)
|
||||
events, err := b.js.Stream(b.ctx, "EVENTS")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var said announce.States
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
select {
|
||||
case err := <-done:
|
||||
t.Fatalf("the assigner stopped: %v", err)
|
||||
default:
|
||||
}
|
||||
m, err := events.GetLastMsgForSubject(b.ctx, announce.Subject("mesh-delivery-assigner"))
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if err := json.Unmarshal(m.Data, &said); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
break
|
||||
}
|
||||
if said.Module != "mesh-delivery-assigner" || said.Node != "n1" || len(said.States) != 3 ||
|
||||
!said.Readable("wants") || !said.Readable("differences") || !said.Readable("compositions") {
|
||||
t.Fatalf("the assigner said %+v, want its three states, each readable", said)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket.
|
||||
func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
|
||||
for _, missing := range readBuckets {
|
||||
b := startBed(t, missing)
|
||||
select {
|
||||
case err := <-b.start(t):
|
||||
if err == nil || !strings.Contains(err.Error(), missing) {
|
||||
t.Errorf("without %s the assigner stopped with %v, want the bucket named", missing, err)
|
||||
}
|
||||
case <-time.After(15 * time.Second):
|
||||
t.Errorf("without %s the assigner started", missing)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// putJSON writes a value into one of the bed's buckets.
|
||||
func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) {
|
||||
t.Helper()
|
||||
if _, err := b.buckets[bucket].Put(b.ctx, key, value); err != nil {
|
||||
t.Fatalf("%s %s: %v", bucket, key, err)
|
||||
}
|
||||
}
|
||||
|
||||
// manifest keeps a manifest in the indexer's bucket by its digest and answers the digest.
|
||||
func (b *bed) manifest(t *testing.T, module, body string) string {
|
||||
t.Helper()
|
||||
digest := delivery.ManifestDigest([]byte(body))
|
||||
key, _ := delivery.ManifestsKey(digest)
|
||||
value, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: body})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.ManifestsBucket, key, value)
|
||||
return digest
|
||||
}
|
||||
|
||||
// 1297/AC1: against the real bus, a fixture change touching one module on one machine is composed and answered within
|
||||
// seconds under the checker's key of the check, naming the machine composed and the mesh-sdk it composed with.
|
||||
func TestAChangeIsComposedOnTheBus(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
trunk := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
changed := b.manifest(t, "web", `{"module":"web","version":"2"}`)
|
||||
mod, err := delivery.EncodeModule(delivery.Module{FP2: delivery.SourcePrefix2 + strings.Repeat("a", 64), Manifest: trunk, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
b.putJSON(t, delivery.ModulesBucket, "web", mod)
|
||||
b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"build-1"}`))
|
||||
b.putJSON(t, provider.AssignmentsBucket, "ace.other", []byte(`{"node":"ace","module":"other","build":"build-2"}`))
|
||||
heads := []delivery.Head{{Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}}
|
||||
hash, _ := delivery.HeadsHash(heads)
|
||||
cand, err := delivery.EncodeCandidate(delivery.Candidate{Group: "feat/x", HeadsHash: hash, Pulls: []delivery.CandidatePull{{Head: heads[0],
|
||||
Modules: []string{"web"}, Fps: map[string]delivery.CandidateModule{"web": {FP2: delivery.SourcePrefix2 + strings.Repeat("b", 64), Manifest: changed}}}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ckey, _ := delivery.CandidatesKey("feat/x")
|
||||
b.putJSON(t, delivery.CandidatesBucket, ckey, cand)
|
||||
began := time.Now()
|
||||
b.start(t)
|
||||
key, _ := delivery.CompositionKey("feat/x", hash)
|
||||
var v delivery.CompositionVerdict
|
||||
for deadline := time.Now().Add(10 * time.Second); ; time.Sleep(50 * time.Millisecond) {
|
||||
if time.Now().After(deadline) {
|
||||
t.Fatalf("no composition under %s after 10 s", key)
|
||||
}
|
||||
e, err := b.buckets[delivery.AssignerCompositionsBucket].Get(b.ctx, key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if v, err = delivery.DecodeCompositionVerdict(e.Value()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
break
|
||||
}
|
||||
if v.State != delivery.CompositionPass || len(v.Machines) != 1 || v.Machines[0] != "g14" || v.SDK == "" {
|
||||
t.Fatalf("composition = %+v, want a pass on g14 alone naming its mesh-sdk", v)
|
||||
}
|
||||
if took := time.Since(began); took > 5*time.Second {
|
||||
t.Errorf("the composition took %s", took)
|
||||
}
|
||||
}
|
||||
|
||||
// get waits for a key of one of the bed's buckets and answers its value.
|
||||
func (b *bed) get(t *testing.T, bucket, key string) []byte {
|
||||
t.Helper()
|
||||
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := b.buckets[bucket].Get(b.ctx, key); err == nil {
|
||||
return e.Value()
|
||||
}
|
||||
}
|
||||
t.Fatalf("no %s %s after 10 s", bucket, key)
|
||||
return nil
|
||||
}
|
||||
|
||||
// 1297/B4, B5 on the bus: the controller sent F to g14 while the canary ace does not run F healthy yet. Started whole,
|
||||
// the assigner wants F on the canary, writes the difference on g14 explained by rule 3, and on ace one unexplained (the
|
||||
// controller sends ace the older fingerprint while ace wants F).
|
||||
func TestWantsAndTheirDifferencesOnTheBus(t *testing.T) {
|
||||
b := startBed(t, "")
|
||||
fNew, fOld := delivery.SourcePrefix2+strings.Repeat("b", 64), delivery.SourcePrefix2+strings.Repeat("a", 64)
|
||||
digest := b.manifest(t, "web", `{"module":"web","version":"1"}`)
|
||||
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fNew, Manifest: digest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
|
||||
b.putJSON(t, delivery.ModulesBucket, "web", mod)
|
||||
built, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "builder.novox.1", Step: "pushed", Manifest: digest})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bkey, _ := delivery.BuilderBuildKey("web", fNew)
|
||||
b.putJSON(t, delivery.BuilderBuildsBucket, bkey, built)
|
||||
gkey, _ := delivery.ControllerGrantKey("web", fNew)
|
||||
grant, _ := delivery.EncodeControllerGrant(delivery.ControllerGrant{})
|
||||
b.putJSON(t, delivery.ControllerGrantsBucket, gkey, grant)
|
||||
for _, node := range []string{"ace", "g14"} {
|
||||
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
|
||||
b.putJSON(t, delivery.PresenceBucket, node, on)
|
||||
have, _ := delivery.EncodeHave(delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
|
||||
b.putJSON(t, delivery.HaveBucket, node+".web", have)
|
||||
}
|
||||
b.putJSON(t, provider.AssignmentsBucket, "ace.web", []byte(`{"node":"ace","module":"web","build":"b-8"}`))
|
||||
b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-9"}`))
|
||||
b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-8"), []byte(`{"id":"b-8","repository":"novox/web","source-fingerprint":"`+fOld+`"}`))
|
||||
b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("web", "b-9"), []byte(`{"id":"b-9","repository":"novox/web","source-fingerprint":"`+fNew+`"}`))
|
||||
b.start(t)
|
||||
|
||||
wkey, _ := delivery.WantKey("ace", "web")
|
||||
want, err := delivery.DecodeWant(b.get(t, delivery.AssignerWantsBucket, wkey))
|
||||
if err != nil || want.FP != fNew {
|
||||
t.Fatalf("the canary ace's want = %+v, %v", want, err)
|
||||
}
|
||||
dkey, _ := delivery.DifferenceKey("g14", "web")
|
||||
d, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, dkey))
|
||||
if err != nil || d.Rule != delivery.RuleCanary || d.Sent != fNew {
|
||||
t.Fatalf("g14's difference = %+v, %v; want rule 3", d, err)
|
||||
}
|
||||
akey, _ := delivery.DifferenceKey("ace", "web")
|
||||
a, err := delivery.DecodeDifference(b.get(t, delivery.AssignerDifferencesBucket, akey))
|
||||
if err != nil || !a.Unexplained || a.Want != fNew || a.Sent != fOld {
|
||||
t.Fatalf("ace's difference = %+v, %v; want it unexplained", a, err)
|
||||
}
|
||||
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, "want.g14.web"); err == nil {
|
||||
t.Errorf("g14, held back by the canary with no want before, was given one: %s", e.Value())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,222 @@
|
||||
// Command mesh-delivery-assigner is delivery's assigner (novox/hq ADR 0429 §1 and §6, ADR 0446, ADR 0528, issue 1297): a
|
||||
// process on its own bus credential that watches what the indexer, the builder, the controller, the node-engines and
|
||||
// the watchman write, and writes what each node should run (`want.<node>.<module>`, by the six rules), the
|
||||
// fingerprints it holds, every difference between a want and what the controller sends, and the composition check's
|
||||
// verdict of each group the checker judges. In release 5 the wants are only compared: the controller still sends what
|
||||
// each node runs (ADR 0429 §7). It holds no timer: every pass is started by a change it watched.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"os"
|
||||
"os/signal"
|
||||
"strings"
|
||||
"syscall"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
if err := run(ctx, os.Getenv, os.Stderr); err != nil {
|
||||
fmt.Fprintln(os.Stderr, "mesh-delivery-assigner:", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
}
|
||||
|
||||
// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
|
||||
// holds the two lists equal), so the controller's broker grants exactly these.
|
||||
var readBuckets = []string{
|
||||
provider.AssignmentsBucket,
|
||||
delivery.BuildsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ManifestsBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
}
|
||||
|
||||
// ownBuckets are the assigner's own states, as its manifest declares them.
|
||||
var ownBuckets = []string{delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket}
|
||||
|
||||
// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
|
||||
// (and the node agent from release 6), its differences by the view, its compositions by the checker.
|
||||
var keptStates = []announce.State{
|
||||
{Name: "compositions", Readable: true},
|
||||
{Name: "differences", Readable: true},
|
||||
{Name: "wants", Readable: true},
|
||||
}
|
||||
|
||||
// buckets is every bucket the assigner opened, by name.
|
||||
type buckets map[string]jetstream.KeyValue
|
||||
|
||||
// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named, since a pass
|
||||
// without it would write wants from half the mesh.
|
||||
func open(ctx context.Context, js jetstream.JetStream) (buckets, error) {
|
||||
out := buckets{}
|
||||
for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
|
||||
kv, err := js.KeyValue(ctx, name)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bucket %s cannot be opened: %w", name, err)
|
||||
}
|
||||
out[name] = kv
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// run is the program: it returns when ctx ends or the bus cannot be used.
|
||||
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
|
||||
logger := log.New(logOut, "mesh-delivery-assigner: ", log.LstdFlags)
|
||||
path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
|
||||
if path == "" {
|
||||
return errors.New("MESH_BROKER_FILE is not set, so there is no credential to read the bus with")
|
||||
}
|
||||
account, err := process.ReadAccount(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
opts := []nats.Option{
|
||||
nats.Name(account.Node + "." + account.Module),
|
||||
nats.UserInfo(account.User, account.Password),
|
||||
nats.CustomInboxPrefix(account.InboxPrefix()),
|
||||
nats.MaxReconnects(-1),
|
||||
nats.RetryOnFailedConnect(true),
|
||||
}
|
||||
if strings.TrimSpace(account.Fingerprint) != "" {
|
||||
opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint)))
|
||||
}
|
||||
conn, err := nats.Connect(account.URL, opts...)
|
||||
if err != nil {
|
||||
return fmt.Errorf("connecting to the bus: %w", err)
|
||||
}
|
||||
defer conn.Close()
|
||||
js, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
kvs, err := open(ctx, js)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot
|
||||
// be said now: process.Conn says it again at every reconnect.
|
||||
said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer said.Close()
|
||||
if err := said.Keeps(keptStates...); err != nil {
|
||||
logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
|
||||
}
|
||||
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
|
||||
return follow(ctx, kvs, logger)
|
||||
}
|
||||
|
||||
// watched are the buckets whose every value a pass reads; the manifests are read by digest when a pass needs one, and
|
||||
// watched only for keys, so a manifest that arrives wakes a composition that waits for it.
|
||||
var watched = []string{
|
||||
provider.AssignmentsBucket,
|
||||
delivery.ControllerGrantsBucket,
|
||||
delivery.BuilderBuildsBucket,
|
||||
delivery.CandidatesBucket,
|
||||
delivery.ModulesBucket,
|
||||
delivery.PresenceBucket,
|
||||
delivery.HaveBucket,
|
||||
delivery.AssignerCompositionsBucket,
|
||||
delivery.AssignerDifferencesBucket,
|
||||
delivery.AssignerWantsBucket,
|
||||
}
|
||||
|
||||
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
|
||||
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
|
||||
func follow(ctx context.Context, kvs buckets, logger *log.Logger) error {
|
||||
w := newWorld()
|
||||
dirty := make(chan struct{}, 1)
|
||||
wake := func() {
|
||||
select {
|
||||
case dirty <- struct{}{}:
|
||||
default:
|
||||
}
|
||||
}
|
||||
errs := make(chan error, len(watched)+1)
|
||||
settled := make(chan struct{}, len(watched)+1)
|
||||
watch := func(name string, opts ...jetstream.WatchOpt) error {
|
||||
kw, err := kvs[name].WatchAll(ctx, opts...)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the bucket %s cannot be watched: %w", name, err)
|
||||
}
|
||||
go func() {
|
||||
defer kw.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
errs <- nil
|
||||
return
|
||||
case e, ok := <-kw.Updates():
|
||||
if !ok {
|
||||
errs <- fmt.Errorf("the watch of %s ended", name)
|
||||
return
|
||||
}
|
||||
if e == nil {
|
||||
settled <- struct{}{}
|
||||
continue
|
||||
}
|
||||
var value []byte
|
||||
if e.Operation() == jetstream.KeyValuePut {
|
||||
value = e.Value()
|
||||
}
|
||||
w.put(name, e.Key(), value)
|
||||
wake()
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
for _, name := range watched {
|
||||
var opts []jetstream.WatchOpt
|
||||
if name == delivery.HaveBucket {
|
||||
opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
|
||||
}
|
||||
if err := watch(name, opts...); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
|
||||
return err
|
||||
}
|
||||
for range len(watched) + 1 {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
case <-settled:
|
||||
}
|
||||
}
|
||||
p := &passer{kvs: kvs, w: w, logf: logger.Printf}
|
||||
wake()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
case <-settled:
|
||||
case <-dirty:
|
||||
p.pass(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
|
||||
// (`<module>.<state>` read as the bucket `<module>_<state>`), so it is granted what it reads and reads nothing else.
|
||||
func TestOpensWhatItsManifestNames(t *testing.T) {
|
||||
raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, err := catalogue.ParseManifest(raw)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var reads []string
|
||||
for _, r := range m.Reads {
|
||||
reads = append(reads, strings.Replace(r, ".", "_", 1))
|
||||
}
|
||||
var states []string
|
||||
for _, s := range m.State {
|
||||
states = append(states, m.Module+"_"+s.Name)
|
||||
}
|
||||
for _, pair := range []struct {
|
||||
what string
|
||||
code, want []string
|
||||
}{{"reads", readBuckets, reads}, {"states", ownBuckets, states}} {
|
||||
got, want := slices.Sorted(slices.Values(pair.code)), slices.Sorted(slices.Values(pair.want))
|
||||
if !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("the program's %s are %v, the manifest's %v", pair.what, got, want)
|
||||
}
|
||||
}
|
||||
var kept []string
|
||||
for _, s := range keptStates {
|
||||
kept = append(kept, m.Module+"_"+s.Name)
|
||||
}
|
||||
if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(ownBuckets)); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("the program announces %v, keeps %v", got, want)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B2a: without a credential the program does not start, and says why.
|
||||
func TestNoCredential(t *testing.T) {
|
||||
err := run(context.Background(), func(string) string { return "" }, io.Discard)
|
||||
if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
|
||||
t.Errorf("run without a credential = %v", err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,207 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/sdkversion"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// passer makes the assigner's passes: from what the world holds now, it writes each key whose value differs from the
|
||||
// one written, and only those (the Same functions).
|
||||
type passer struct {
|
||||
kvs buckets
|
||||
w *world
|
||||
logf func(string, ...any)
|
||||
sdk string
|
||||
}
|
||||
|
||||
func (p *passer) pass(ctx context.Context) {
|
||||
p.compositions(ctx)
|
||||
rules := p.wants(ctx)
|
||||
p.differences(ctx, rules)
|
||||
}
|
||||
|
||||
// differences writes, for each node and module the assigner wants or the controller sends, the difference between the
|
||||
// two with the rule that explains it or unexplained (ADR 0528 clause 4), and deletes the key of a difference that
|
||||
// ended; its history keeps what it was.
|
||||
func (p *passer) differences(ctx context.Context, rules map[string]map[string]assign.Holding) {
|
||||
_, assignments := p.w.assigned()
|
||||
wanted := p.w.wantsOf()
|
||||
trunk := p.w.trunk()
|
||||
pairs := map[[2]string]bool{}
|
||||
for _, a := range assignments {
|
||||
pairs[[2]string{a.Node, a.Module}] = true
|
||||
}
|
||||
for module, byNode := range wanted {
|
||||
for node := range byNode {
|
||||
pairs[[2]string{node, module}] = true
|
||||
}
|
||||
}
|
||||
written := p.w.read(delivery.AssignerDifferencesBucket)
|
||||
keep := map[string]bool{}
|
||||
for pair := range pairs {
|
||||
node, module := pair[0], pair[1]
|
||||
key, err := delivery.DifferenceKey(node, module)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
var want *delivery.Want
|
||||
if w, ok := wanted[module][node]; ok {
|
||||
want = &w
|
||||
}
|
||||
var sent assign.Sent
|
||||
if a, ok := assignments[node+"."+module]; ok {
|
||||
sent = assign.Sent{Sends: true, Build: a.Build, FP: p.w.sentFP(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module, a.Build)}
|
||||
}
|
||||
d, ok := assign.Compare(want, sent, trunk[module].FP2, rules[module][node])
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
keep[key] = true
|
||||
if before, err := delivery.DecodeDifference(written[key]); err == nil {
|
||||
before.V = 0
|
||||
if delivery.SameDifference(before, d) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
p.put(ctx, delivery.AssignerDifferencesBucket, key, func() ([]byte, error) { return delivery.EncodeDifference(d) })
|
||||
}
|
||||
for key := range written {
|
||||
if keep[key] {
|
||||
continue
|
||||
}
|
||||
if err := p.kvs[delivery.AssignerDifferencesBucket].Delete(ctx, key); err != nil {
|
||||
p.logf("the difference %s that ended could not be deleted: %v", key, err)
|
||||
continue
|
||||
}
|
||||
p.w.put(delivery.AssignerDifferencesBucket, key, nil)
|
||||
}
|
||||
}
|
||||
|
||||
// wants writes, for every module the trunk holds a fingerprint of, each node's want by the six rules (ADR 0429 §6), and
|
||||
// the hold of a fingerprint whose health check failed. It returns, by module then node, the rule that holds a node
|
||||
// back, for the comparison.
|
||||
func (p *passer) wants(ctx context.Context) map[string]map[string]assign.Holding {
|
||||
get := kvGetter(p.kvs[delivery.ManifestsBucket])
|
||||
assigned, _ := p.w.assigned()
|
||||
trunk := p.w.trunk()
|
||||
manifests := map[string]catalogue.Manifest{}
|
||||
for name, mod := range trunk {
|
||||
if mod.Manifest == "" {
|
||||
continue
|
||||
}
|
||||
if got := p.w.manifest(ctx, get, mod.Manifest); got.err == "" {
|
||||
manifests[name] = got.manifest
|
||||
}
|
||||
}
|
||||
nodes := p.w.nodes(manifests, assigned)
|
||||
nodesOf := map[string][]string{}
|
||||
for node, modules := range assigned {
|
||||
for _, m := range modules {
|
||||
nodesOf[m] = append(nodesOf[m], node)
|
||||
}
|
||||
}
|
||||
wanted := p.w.wantsOf()
|
||||
rules := map[string]map[string]assign.Holding{}
|
||||
names := map[string]bool{}
|
||||
for name := range trunk {
|
||||
names[name] = true
|
||||
}
|
||||
for name := range wanted {
|
||||
names[name] = true
|
||||
}
|
||||
for name := range names {
|
||||
fp := trunk[name].FP2
|
||||
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name]}
|
||||
if fp != "" {
|
||||
held, _ := delivery.HeldKey(name, fp)
|
||||
grant, _ := delivery.ControllerGrantKey(name, fp)
|
||||
m.Built, m.Held, m.Granted = p.w.built(name, fp), p.w.has(delivery.AssignerWantsBucket, held), p.w.has(delivery.ControllerGrantsBucket, grant)
|
||||
}
|
||||
last := map[string]string{}
|
||||
for _, node := range m.Nodes {
|
||||
last[node] = p.w.lastHealthy(node, name, fp)
|
||||
}
|
||||
out := assign.Wants(m, nodes, wanted[name], last)
|
||||
rules[name] = out.Rules
|
||||
if out.Held != nil {
|
||||
key, err := delivery.HeldKey(name, fp)
|
||||
if err == nil {
|
||||
held := *out.Held
|
||||
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeHeld(held) })
|
||||
}
|
||||
}
|
||||
for node, want := range out.Wants {
|
||||
key, err := delivery.WantKey(node, name)
|
||||
if err != nil {
|
||||
p.logf("no want of %s on %s is written: %v", name, node, err)
|
||||
continue
|
||||
}
|
||||
want := want
|
||||
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeWant(want) })
|
||||
}
|
||||
}
|
||||
return rules
|
||||
}
|
||||
|
||||
// compositions writes the composition check's verdict of every group the indexer holds a candidate for, under the
|
||||
// checker's key of that check: waits, with why, while the indexer holds what it composes from only in part.
|
||||
func (p *passer) compositions(ctx context.Context) {
|
||||
if p.sdk == "" {
|
||||
p.sdk = sdkversion.Version()
|
||||
}
|
||||
written := p.w.read(delivery.AssignerCompositionsBucket)
|
||||
trunk := p.w.trunk()
|
||||
var mesh *assign.Mesh
|
||||
for key, value := range p.w.read(delivery.CandidatesBucket) {
|
||||
c, err := delivery.DecodeCandidate(value)
|
||||
if err != nil {
|
||||
p.logf("the candidate %s cannot be read, so it is not composed: %v", key, err)
|
||||
continue
|
||||
}
|
||||
out, err := delivery.CompositionKey(c.Group, c.HeadsHash)
|
||||
if err != nil {
|
||||
p.logf("the candidate %s names no composition key: %v", key, err)
|
||||
continue
|
||||
}
|
||||
began := time.Now()
|
||||
ch, waits := p.w.change(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]), c, trunk)
|
||||
var v delivery.CompositionVerdict
|
||||
if waits != "" {
|
||||
v = delivery.CompositionVerdict{Group: c.Group, Heads: c.HeadsHash, State: delivery.CompositionWaits, Words: waits}
|
||||
} else {
|
||||
if mesh == nil {
|
||||
m := p.w.mesh(ctx, kvGetter(p.kvs[delivery.ManifestsBucket]))
|
||||
mesh = &m
|
||||
}
|
||||
v = assign.Compose(*mesh, ch, p.sdk, time.Since(began))
|
||||
}
|
||||
if before, err := delivery.DecodeCompositionVerdict(written[out]); err == nil {
|
||||
before.Took, before.V = v.Took, 0
|
||||
if delivery.SameCompositionVerdict(before, v) {
|
||||
continue
|
||||
}
|
||||
}
|
||||
p.put(ctx, delivery.AssignerCompositionsBucket, out, func() ([]byte, error) { return delivery.EncodeCompositionVerdict(v) })
|
||||
}
|
||||
}
|
||||
|
||||
// put writes one key of the assigner's own bucket; a write that fails is said and made again at the next pass.
|
||||
func (p *passer) put(ctx context.Context, bucket, key string, encode func() ([]byte, error)) {
|
||||
value, err := encode()
|
||||
if err != nil {
|
||||
p.logf("%s %s is not written: %v", bucket, key, err)
|
||||
return
|
||||
}
|
||||
if _, err := p.kvs[bucket].Put(ctx, key, value); err != nil && !errors.Is(err, context.Canceled) {
|
||||
p.logf("%s %s could not be written: %v", bucket, key, err)
|
||||
return
|
||||
}
|
||||
p.w.put(bucket, key, value)
|
||||
}
|
||||
@@ -0,0 +1,370 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// noModuleWords is the indexer's reason for a module of a candidate whose directory the merge preview holds no manifest
|
||||
// for (mesh-sdk delivery.CandidateNoneNoModule, which the indexer writes): the change removes the module.
|
||||
const noModuleWords = delivery.CandidateNoneNoModule
|
||||
|
||||
// assignment holds only the fields of the controller's Assignment the assigner reads (the view's own, mesh-delivery-view
|
||||
// internal/view/view.go): an assignment stands while it has not ended and has no state.
|
||||
type assignment struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Build string `json:"build"`
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
|
||||
func (a assignment) open() bool { return a.Ended.IsZero() && a.State == "" }
|
||||
|
||||
// world is what the assigner watched, as it stands now: every value by bucket and key, and the manifests read by
|
||||
// digest. The watches write it; a pass reads it.
|
||||
type world struct {
|
||||
mu sync.Mutex
|
||||
values map[string]map[string][]byte
|
||||
manifests map[string]parsed
|
||||
// healthy is, by have key, every fingerprint the node ran healthy, oldest first, each once: what a set-back goes to.
|
||||
healthy map[string][]string
|
||||
// sent is, by build id, the fingerprint of the new form the controller's build record names, once it names one.
|
||||
sent map[string]string
|
||||
}
|
||||
|
||||
// parsed is a manifest read by digest: its module and its manifest, or why it cannot be read.
|
||||
type parsed struct {
|
||||
module string
|
||||
manifest catalogue.Manifest
|
||||
err string
|
||||
}
|
||||
|
||||
func newWorld() *world {
|
||||
return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, sent: map[string]string{}}
|
||||
}
|
||||
|
||||
// put takes a change of a watched bucket in: a value, or nil for a key deleted or purged.
|
||||
func (w *world) put(bucket, key string, value []byte) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if w.values[bucket] == nil {
|
||||
w.values[bucket] = map[string][]byte{}
|
||||
}
|
||||
if value == nil {
|
||||
delete(w.values[bucket], key)
|
||||
return
|
||||
}
|
||||
w.values[bucket][key] = value
|
||||
if bucket == delivery.HaveBucket {
|
||||
if h, err := delivery.DecodeHave(value); err == nil && h.FP != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK {
|
||||
list := slices.DeleteFunc(w.healthy[key], func(fp string) bool { return fp == h.FP })
|
||||
w.healthy[key] = append(list, h.FP)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// lastHealthy is the newest fingerprint other than fp the node ran the module healthy at, or empty.
|
||||
func (w *world) lastHealthy(node, module, fp string) string {
|
||||
key, err := delivery.HaveKey(node, module)
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
list := w.healthy[key]
|
||||
for i := len(list) - 1; i >= 0; i-- {
|
||||
if list[i] != fp {
|
||||
return list[i]
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// read is a copy of what a bucket holds now.
|
||||
func (w *world) read(bucket string) map[string][]byte {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
out := make(map[string][]byte, len(w.values[bucket]))
|
||||
for k, v := range w.values[bucket] {
|
||||
out[k] = v
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// getter reads one key's value; errNotThere when the key does not exist.
|
||||
type getter func(ctx context.Context, key string) ([]byte, error)
|
||||
|
||||
var errNotThere = errors.New("no such key")
|
||||
|
||||
// kvGetter reads keys of a bucket.
|
||||
func kvGetter(kv jetstream.KeyValue) getter {
|
||||
return func(ctx context.Context, key string) ([]byte, error) {
|
||||
e, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil, errNotThere
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return e.Value(), nil
|
||||
}
|
||||
}
|
||||
|
||||
// manifest reads the manifest of a digest from the indexer's bucket once and keeps it: a digest's manifest never changes.
|
||||
func (w *world) manifest(ctx context.Context, get getter, digest string) parsed {
|
||||
w.mu.Lock()
|
||||
p, ok := w.manifests[digest]
|
||||
w.mu.Unlock()
|
||||
if ok {
|
||||
return p
|
||||
}
|
||||
key, err := delivery.ManifestsKey(digest)
|
||||
if err != nil {
|
||||
return parsed{err: err.Error()}
|
||||
}
|
||||
value, err := get(ctx, key)
|
||||
if errors.Is(err, errNotThere) {
|
||||
return parsed{err: "the indexer holds no manifest of digest " + digest + " yet"}
|
||||
}
|
||||
if err != nil {
|
||||
return parsed{err: err.Error()}
|
||||
}
|
||||
m, err := delivery.DecodeManifest(value)
|
||||
if err != nil {
|
||||
p = parsed{err: err.Error()}
|
||||
} else if man, err := catalogue.ParseManifest([]byte(m.JSON)); err != nil {
|
||||
p = parsed{module: m.Module, err: err.Error()}
|
||||
} else {
|
||||
p = parsed{module: m.Module, manifest: man}
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.manifests[digest] = p
|
||||
w.mu.Unlock()
|
||||
return p
|
||||
}
|
||||
|
||||
// assigned is each node's modules with an open assignment, read from the controller's assignments, in name order.
|
||||
func (w *world) assigned() (map[string][]string, map[string]assignment) {
|
||||
out, all := map[string][]string{}, map[string]assignment{}
|
||||
for _, value := range w.read(provider.AssignmentsBucket) {
|
||||
var a assignment
|
||||
if json.Unmarshal(value, &a) != nil || a.Node == "" || a.Module == "" || !a.open() {
|
||||
continue
|
||||
}
|
||||
out[a.Node] = append(out[a.Node], a.Module)
|
||||
all[a.Node+"."+a.Module] = a
|
||||
}
|
||||
for _, mods := range out {
|
||||
sort.Strings(mods)
|
||||
}
|
||||
return out, all
|
||||
}
|
||||
|
||||
// trunk is every module the indexer holds at the trunk, with its value.
|
||||
func (w *world) trunk() map[string]delivery.Module {
|
||||
out := map[string]delivery.Module{}
|
||||
for key, value := range w.read(delivery.ModulesBucket) {
|
||||
name, err := delivery.ParseModulesKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if m, err := delivery.DecodeModule(value); err == nil {
|
||||
out[name] = m
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// mesh is the mesh as the trunk holds it, for the composition check.
|
||||
func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
|
||||
m := assign.Mesh{Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}}
|
||||
m.Assigned, _ = w.assigned()
|
||||
for name, mod := range w.trunk() {
|
||||
if mod.Manifest == "" {
|
||||
continue
|
||||
}
|
||||
p := w.manifest(ctx, manifests, mod.Manifest)
|
||||
if p.err != "" {
|
||||
m.Unreadable[name] = p.err
|
||||
continue
|
||||
}
|
||||
m.Manifests[name] = p.manifest
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// change is a candidate's change for the composition check, or why the composition waits for it.
|
||||
func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) {
|
||||
ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, Twice: map[string][]string{}}
|
||||
definedIn := map[string]map[string]bool{}
|
||||
for _, p := range c.Pulls {
|
||||
if p.Repository == assign.SDKRepository {
|
||||
ch.SDKHead = p.Head.Head
|
||||
}
|
||||
for _, name := range p.Modules {
|
||||
fp, ok := p.Fps[name]
|
||||
if !ok || fp.Manifest == "" {
|
||||
if ok && fp.None2 == noModuleWords {
|
||||
ch.Removed = append(ch.Removed, name)
|
||||
continue
|
||||
}
|
||||
return ch, fmt.Sprintf("the indexer holds no manifest of %s for %s#%d yet", name, p.Repository, p.Number)
|
||||
}
|
||||
if fp.FP2 != "" && fp.FP2 != trunk[name].FP2 {
|
||||
ch.Rebuilt++
|
||||
}
|
||||
got := w.manifest(ctx, manifests, fp.Manifest)
|
||||
if strings.HasPrefix(got.err, "the indexer holds no manifest") {
|
||||
return ch, got.err
|
||||
}
|
||||
if got.err != "" {
|
||||
ch.Unreadable[name] = got.err
|
||||
continue
|
||||
}
|
||||
ch.Manifests[got.module] = got.manifest
|
||||
if definedIn[got.module] == nil {
|
||||
definedIn[got.module] = map[string]bool{}
|
||||
}
|
||||
definedIn[got.module][p.Repository] = true
|
||||
if t, ok := trunk[got.module]; ok && t.Repository != "" && t.Repository != p.Repository {
|
||||
definedIn[got.module][t.Repository] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
for name, repos := range definedIn {
|
||||
if len(repos) > 1 {
|
||||
for r := range repos {
|
||||
ch.Twice[name] = append(ch.Twice[name], r)
|
||||
}
|
||||
sort.Strings(ch.Twice[name])
|
||||
}
|
||||
}
|
||||
return ch, ""
|
||||
}
|
||||
|
||||
// wantsOf is the wants the assigner wrote, by module then node.
|
||||
func (w *world) wantsOf() map[string]map[string]delivery.Want {
|
||||
out := map[string]map[string]delivery.Want{}
|
||||
for key, value := range w.read(delivery.AssignerWantsBucket) {
|
||||
node, module, err := delivery.ParseWantKey(key)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
want, err := delivery.DecodeWant(value)
|
||||
if err != nil {
|
||||
continue
|
||||
}
|
||||
if out[module] == nil {
|
||||
out[module] = map[string]delivery.Want{}
|
||||
}
|
||||
out[module][node] = want
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// nodes is what the six rules read of the nodes: presence, haves, and every provision's providers with their nodes.
|
||||
func (w *world) nodes(manifests map[string]catalogue.Manifest, assigned map[string][]string) assign.Nodes {
|
||||
n := assign.Nodes{Online: map[string]bool{}, Haves: map[string]map[string]delivery.Have{}, Providers: map[string]map[string][]string{}}
|
||||
for key, value := range w.read(delivery.PresenceBucket) {
|
||||
node, _, _, err := delivery.ParsePresenceKey(key)
|
||||
if err != nil || node == "" {
|
||||
continue
|
||||
}
|
||||
if p, err := delivery.DecodePresence(value); err == nil {
|
||||
n.Online[node] = p.Online
|
||||
}
|
||||
}
|
||||
for key, value := range w.read(delivery.HaveBucket) {
|
||||
node, module, isNode, err := delivery.ParseHaveKey(key)
|
||||
if err != nil || isNode {
|
||||
continue
|
||||
}
|
||||
if h, err := delivery.DecodeHave(value); err == nil {
|
||||
if n.Haves[node] == nil {
|
||||
n.Haves[node] = map[string]delivery.Have{}
|
||||
}
|
||||
n.Haves[node][module] = h
|
||||
}
|
||||
}
|
||||
nodesOf := map[string][]string{}
|
||||
for node, modules := range assigned {
|
||||
for _, m := range modules {
|
||||
nodesOf[m] = append(nodesOf[m], node)
|
||||
}
|
||||
}
|
||||
for name, man := range manifests {
|
||||
for _, o := range man.Provides {
|
||||
if n.Providers[o.Name] == nil {
|
||||
n.Providers[o.Name] = map[string][]string{}
|
||||
}
|
||||
n.Providers[o.Name][name] = nodesOf[name]
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// has says whether a bucket holds a key now.
|
||||
func (w *world) has(bucket, key string) bool {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
_, ok := w.values[bucket][key]
|
||||
return ok
|
||||
}
|
||||
|
||||
// built says whether the builder's build of a module at a fingerprint is done.
|
||||
func (w *world) built(module, fp string) bool {
|
||||
key, err := delivery.BuilderBuildKey(module, fp)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
w.mu.Lock()
|
||||
value, ok := w.values[delivery.BuilderBuildsBucket][key]
|
||||
w.mu.Unlock()
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
b, err := delivery.DecodeBuilderBuild(value)
|
||||
return err == nil && b.State == delivery.BuildDone
|
||||
}
|
||||
|
||||
// sentFP is the fingerprint of the new form of the build the controller's assignment names, read from its build record
|
||||
// (mesh-controller_builds `<module>.build.<id>`) once and kept; empty when the record names none.
|
||||
func (w *world) sentFP(ctx context.Context, get getter, module, build string) string {
|
||||
if build == "" {
|
||||
return ""
|
||||
}
|
||||
w.mu.Lock()
|
||||
fp, ok := w.sent[build]
|
||||
w.mu.Unlock()
|
||||
if ok {
|
||||
return fp
|
||||
}
|
||||
value, err := get(ctx, delivery.BuildKey(module, build))
|
||||
if err != nil {
|
||||
return ""
|
||||
}
|
||||
var rec delivery.BuildRecord
|
||||
if json.Unmarshal(value, &rec) != nil || !strings.HasPrefix(rec.SourceFingerprint, delivery.SourcePrefix2) {
|
||||
return ""
|
||||
}
|
||||
w.mu.Lock()
|
||||
w.sent[build] = rec.SourceFingerprint
|
||||
w.mu.Unlock()
|
||||
return rec.SourceFingerprint
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-assigner/internal/assign"
|
||||
)
|
||||
|
||||
// fakeManifests is the indexer's manifests bucket in memory.
|
||||
type fakeManifests map[string][]byte
|
||||
|
||||
func (f fakeManifests) add(t *testing.T, module, body string) string {
|
||||
t.Helper()
|
||||
digest := delivery.ManifestDigest([]byte(body))
|
||||
value, err := delivery.EncodeManifest(delivery.Manifest{Module: module, JSON: body})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
f[digest] = value
|
||||
return digest
|
||||
}
|
||||
|
||||
func (f fakeManifests) get(_ context.Context, key string) ([]byte, error) {
|
||||
if v, ok := f[key]; ok {
|
||||
return v, nil
|
||||
}
|
||||
return nil, errNotThere
|
||||
}
|
||||
|
||||
func fp(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) }
|
||||
|
||||
func pull(repo string, number int, modules map[string]delivery.CandidateModule) delivery.CandidatePull {
|
||||
p := delivery.CandidatePull{Head: delivery.Head{Repository: repo, Number: number, Head: strings.Repeat("3", 40)}, Fps: modules}
|
||||
for m := range modules {
|
||||
p.Modules = append(p.Modules, m)
|
||||
}
|
||||
slices.Sort(p.Modules)
|
||||
return p
|
||||
}
|
||||
|
||||
// 1297/B3b: a candidate's change reads each module's manifest by its digest, counts the modules rebuilt, takes a
|
||||
// module with no manifest in the preview as removed, a mesh-sdk member's head, and a module the trunk holds from
|
||||
// another repository as defined twice.
|
||||
func TestTheChangeOfACandidate(t *testing.T) {
|
||||
f := fakeManifests{}
|
||||
web := f.add(t, "web", `{"module":"web","version":"2"}`)
|
||||
mail := f.add(t, "mail", `{"module":"mail","version":"1"}`)
|
||||
trunk := map[string]delivery.Module{"web": {FP2: fp('a'), Repository: "novox/web"}, "mail": {FP2: fp('c'), Repository: "novox/mail"}}
|
||||
c := delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: web}, "old": {None2: noModuleWords}}),
|
||||
pull("novox/other", 2, map[string]delivery.CandidateModule{"mail": {FP2: fp('c'), Manifest: mail}}),
|
||||
pull(assign.SDKRepository, 3, nil),
|
||||
}}
|
||||
ch, waits := newWorld().change(context.Background(), f.get, c, trunk)
|
||||
if waits != "" {
|
||||
t.Fatalf("the change waits: %s", waits)
|
||||
}
|
||||
if ch.Manifests["web"].Version != "2" || ch.Rebuilt != 1 || !slices.Equal(ch.Removed, []string{"old"}) || ch.SDKHead != strings.Repeat("3", 40) {
|
||||
t.Errorf("change = %+v", ch)
|
||||
}
|
||||
if !slices.Equal(ch.Twice["mail"], []string{"novox/mail", "novox/other"}) || len(ch.Twice) != 1 {
|
||||
t.Errorf("twice = %v, want mail in novox/mail and novox/other", ch.Twice)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3b: a candidate whose module has no manifest digest yet, or whose manifest the indexer does not hold yet, waits,
|
||||
// saying for what; an unreadable manifest is the change's, not a wait.
|
||||
func TestAChangeWaitsForItsManifests(t *testing.T) {
|
||||
f := fakeManifests{}
|
||||
bad := f.add(t, "web", `{"module":"web","version":"2","scope":"x"}`)
|
||||
for _, c := range []delivery.CandidatePull{
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {None2: "no preview yet"}}),
|
||||
pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: strings.Repeat("9", 64)}}),
|
||||
} {
|
||||
_, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab", Pulls: []delivery.CandidatePull{c}}, nil)
|
||||
if waits == "" {
|
||||
t.Errorf("a change without its manifest %+v does not wait", c.Fps)
|
||||
}
|
||||
}
|
||||
ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab",
|
||||
Pulls: []delivery.CandidatePull{pull("novox/web", 1, map[string]delivery.CandidateModule{"web": {FP2: fp('b'), Manifest: bad}})}}, nil)
|
||||
if waits != "" || ch.Unreadable["web"] == "" {
|
||||
t.Errorf("an unreadable manifest: waits %q, change %+v", waits, ch)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
module mesh-delivery-assigner
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
@@ -0,0 +1,27 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
|
||||
github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
|
||||
github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
|
||||
github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
|
||||
github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc=
|
||||
github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
|
||||
github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
|
||||
github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
|
||||
github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
|
||||
github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
|
||||
github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
|
||||
golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
|
||||
golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
|
||||
golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
|
||||
golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
|
||||
golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
|
||||
golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE=
|
||||
golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s=
|
||||
@@ -0,0 +1,52 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Sent is what the controller sends a node of a module: whether it sends it at all (an open assignment), the build its
|
||||
// assignment names, and that build's fingerprint of the new form (empty when its record names none).
|
||||
type Sent struct {
|
||||
Sends bool
|
||||
Build string
|
||||
FP string
|
||||
}
|
||||
|
||||
// Compare is the difference between a node's want for a module and what the controller sends it (ADR 0528 clause 4),
|
||||
// and whether there is one. want is nil when the assigner wrote none; newest is the module's newest fingerprint and
|
||||
// hold the rule that holds the node back from it (Rule 0 when none does). A difference is explained by a rule when the
|
||||
// controller sent the newest fingerprint while that rule holds the node's want back; every other is unexplained, said
|
||||
// in words, for acceptance test 17 of design 47 to read.
|
||||
func Compare(want *delivery.Want, sent Sent, newest string, hold Holding) (delivery.Difference, bool) {
|
||||
var d delivery.Difference
|
||||
removed := want != nil && want.Removed
|
||||
if want != nil && !removed {
|
||||
d.Want = want.FP
|
||||
}
|
||||
d.WantRemoved = removed
|
||||
if !sent.Sends {
|
||||
if want == nil || removed {
|
||||
return delivery.Difference{}, false
|
||||
}
|
||||
d.Unexplained, d.Words = true, "the node wants "+short(want.FP)+" and the controller sends it nothing"
|
||||
return d, true
|
||||
}
|
||||
d.Sent, d.SentBuild = sent.FP, sent.Build
|
||||
switch {
|
||||
case sent.FP == "":
|
||||
d.Unexplained, d.Words = true, fmt.Sprintf("the build %s the controller sends names no fingerprint of the new form", sent.Build)
|
||||
case removed:
|
||||
d.Unexplained, d.Words = true, fmt.Sprintf("the want is removed and the controller still sends %s", short(sent.FP))
|
||||
case sent.FP == d.Want:
|
||||
return delivery.Difference{}, false
|
||||
case hold.Rule != 0 && sent.FP == newest:
|
||||
d.Rule, d.Words = hold.Rule, fmt.Sprintf("the controller sent %s while %s", short(sent.FP), hold.Words)
|
||||
case d.Want == "":
|
||||
d.Unexplained, d.Words = true, fmt.Sprintf("the controller sends %s and the node has no want", short(sent.FP))
|
||||
default:
|
||||
d.Unexplained, d.Words = true, fmt.Sprintf("the controller sends %s and the node wants %s", short(sent.FP), short(d.Want))
|
||||
}
|
||||
return d, true
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// 1297/B5: given the controller sent F to a node while the canary is not yet healthy on F, the difference names rule 3.
|
||||
func TestADifferenceExplainedByTheCanary(t *testing.T) {
|
||||
m, n := ready()
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
|
||||
out := Wants(m, n, wantedOld(), nil)
|
||||
want := delivery.Want{FP: fOld, Newest: fNew, Rule: delivery.RuleCanary}
|
||||
d, ok := Compare(&want, Sent{Sends: true, Build: "b-9", FP: fNew}, fNew, out.Rules["g14"])
|
||||
if !ok || d.Rule != delivery.RuleCanary || d.Unexplained || d.Want != fOld || d.Sent != fNew {
|
||||
t.Fatalf("difference = %+v, %v; want rule 3", d, ok)
|
||||
}
|
||||
if _, err := delivery.EncodeDifference(d); err != nil {
|
||||
t.Errorf("the difference cannot be written: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B5: a want equal to what is sent is no difference; nothing sent and no want is none; every other is written
|
||||
// unexplained, with words, and is writable.
|
||||
func TestCompare(t *testing.T) {
|
||||
w := delivery.Want{FP: fNew}
|
||||
if _, ok := Compare(&w, Sent{Sends: true, Build: "b", FP: fNew}, fNew, Holding{}); ok {
|
||||
t.Errorf("an equal want and send is a difference")
|
||||
}
|
||||
if _, ok := Compare(nil, Sent{}, fNew, Holding{}); ok {
|
||||
t.Errorf("no want and nothing sent is a difference")
|
||||
}
|
||||
removed := delivery.Want{Removed: true}
|
||||
if _, ok := Compare(&removed, Sent{}, fNew, Holding{}); ok {
|
||||
t.Errorf("a removed want and nothing sent is a difference")
|
||||
}
|
||||
old := delivery.Want{FP: fOld}
|
||||
for name, c := range map[string]struct {
|
||||
want *delivery.Want
|
||||
sent Sent
|
||||
}{
|
||||
"sends an older one": {&w, Sent{Sends: true, Build: "b", FP: fOld}},
|
||||
"names no fingerprint": {&w, Sent{Sends: true, Build: "b"}},
|
||||
"sends a removed module": {&removed, Sent{Sends: true, Build: "b", FP: fNew}},
|
||||
"sends nothing": {&w, Sent{}},
|
||||
"no want": {nil, Sent{Sends: true, Build: "b", FP: fNew}},
|
||||
"no rule holds it back": {&old, Sent{Sends: true, Build: "b", FP: fNew}},
|
||||
} {
|
||||
d, ok := Compare(c.want, c.sent, fNew, Holding{})
|
||||
if !ok || !d.Unexplained || d.Words == "" {
|
||||
t.Errorf("%s: difference = %+v, %v; want it unexplained, with words", name, d, ok)
|
||||
continue
|
||||
}
|
||||
if _, err := delivery.EncodeDifference(d); err != nil {
|
||||
t.Errorf("%s: the difference cannot be written: %v", name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,196 @@
|
||||
// Package assign is the assigner's judgement, as functions over what it read from the bus: the composition check of a
|
||||
// change (novox/hq ADR 0446 clauses 1 to 5, ADR 0528 clause 5, issue 1297), the six rules of what each node should
|
||||
// run (ADR 0429 §6, ADR 0528 clause 3) and the comparison with what the controller sends (ADR 0528 clause 4). It reads
|
||||
// and writes nothing itself: the program hands it what it watched and writes what it answers.
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/compose"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// SDKRepository is the repository of mesh-sdk: a group with a member there changes the composition itself.
|
||||
const SDKRepository = "novox/mesh-sdk"
|
||||
|
||||
// NotJudged names, in a composition's warnings, the failures of ADR 0446 clause 1 the assigner cannot judge from the
|
||||
// bus, because what they need is only in the controller's facts snapshot, which is not on the bus yet (issue 1461;
|
||||
// the design limit of R4g, issue 1297): said on every verdict, never passed silently.
|
||||
const NotJudged = "not judged until the facts snapshot is on the bus (issue 1461): a consumer left out of a grant, a " +
|
||||
"credential left bound elsewhere, a stored setting the changed definition cannot keep, and a new module the " +
|
||||
"node-engine's validator would refuse"
|
||||
|
||||
// Mesh is the mesh as the trunk holds it: every module's manifest by name, the manifests that cannot be read with
|
||||
// why, and each node's open assignments.
|
||||
type Mesh struct {
|
||||
Manifests map[string]catalogue.Manifest
|
||||
Unreadable map[string]string
|
||||
Assigned map[string][]string
|
||||
}
|
||||
|
||||
// Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those
|
||||
// it cannot read and why, the modules it removes, every module defined twice with where, the heads of a mesh-sdk
|
||||
// member (empty when none), and how many modules it rebuilds.
|
||||
type Change struct {
|
||||
Group, Heads string
|
||||
Manifests map[string]catalogue.Manifest
|
||||
Unreadable map[string]string
|
||||
Removed []string
|
||||
Twice map[string][]string
|
||||
SDKHead string
|
||||
Rebuilt int
|
||||
}
|
||||
|
||||
// Touched is the machines a change touches (ADR 0446 clause 3): those running a module the change moves, adds or
|
||||
// removes, those running a consumer of a provision such a module provides (before or after), and those running a
|
||||
// provider of a provision such a module requires after the change. A module the change adds that is assigned nowhere
|
||||
// touches no machine.
|
||||
func Touched(m Mesh, c Change) []string {
|
||||
moved := map[string]bool{}
|
||||
for name := range c.Manifests {
|
||||
moved[name] = true
|
||||
}
|
||||
for name := range c.Unreadable {
|
||||
moved[name] = true
|
||||
}
|
||||
for _, name := range c.Removed {
|
||||
moved[name] = true
|
||||
}
|
||||
provided, required := map[string]bool{}, map[string]bool{}
|
||||
for name := range moved {
|
||||
for _, man := range []catalogue.Manifest{m.Manifests[name], c.Manifests[name]} {
|
||||
for _, o := range man.Provides {
|
||||
provided[o.Name] = true
|
||||
}
|
||||
}
|
||||
for _, r := range c.Manifests[name].Requires {
|
||||
required[r] = true
|
||||
}
|
||||
}
|
||||
var out []string
|
||||
for node, modules := range m.Assigned {
|
||||
for _, mod := range modules {
|
||||
man := m.Manifests[mod]
|
||||
if moved[mod] || slices.ContainsFunc(man.Requires, func(r string) bool { return provided[r] }) ||
|
||||
slices.ContainsFunc(man.Provides, func(o catalogue.Offer) bool { return required[o.Name] }) {
|
||||
out = append(out, node)
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// input is the composition's input for the mesh's catalogue and assignments: the nodes by name with what is assigned
|
||||
// to them, and every seat the assigned modules claim as held there. What the bus does not carry (places on the
|
||||
// private network, settings, credentials, licences) is left empty, so a trunk composition and a change's are made
|
||||
// with the same gaps and compared (ADR 0446 clause 1: what was already broken is said and fails nothing).
|
||||
func input(catalog map[string]catalogue.Manifest, assigned map[string][]string) compose.Input {
|
||||
in := compose.Input{Catalogue: catalog}
|
||||
nodes := make([]string, 0, len(assigned))
|
||||
for n := range assigned {
|
||||
nodes = append(nodes, n)
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
for _, n := range nodes {
|
||||
in.Nodes = append(in.Nodes, compose.Node{Name: n, Assigned: slices.Clone(assigned[n])})
|
||||
for _, mod := range assigned[n] {
|
||||
for _, cl := range catalog[mod].Claims {
|
||||
scope := cl.Scope
|
||||
if seat, ok := catalogue.SeatNamed(cl.Name); ok && scope == "" {
|
||||
scope = seat.Scope
|
||||
}
|
||||
in.Holdings = append(in.Holdings, catalogue.Held{Claim: cl.Name, Scope: scope, Node: n, Module: mod})
|
||||
}
|
||||
}
|
||||
}
|
||||
return in
|
||||
}
|
||||
|
||||
// Compose is the composition check of change c over the mesh m, composed with the mesh-sdk version sdk: every touched
|
||||
// machine resolved as the trunk holds it and as the change leaves it. It fails on a manifest of the change that cannot
|
||||
// be read, a module a machine runs that the change removes, a module defined twice, and a machine that resolves on the
|
||||
// trunk and not with the change; a machine that does not resolve on either is said and fails nothing. It warns past
|
||||
// CompositionRebuildWarning modules rebuilt, says what it does not judge (NotJudged), and for a group with a mesh-sdk
|
||||
// member says it composed with its own mesh-sdk, not the member's head (issue 1621). took is how long it took.
|
||||
func Compose(m Mesh, c Change, sdk string, took time.Duration) delivery.CompositionVerdict {
|
||||
v := delivery.CompositionVerdict{Group: c.Group, Heads: c.Heads, SDK: sdk, Took: took.Milliseconds()}
|
||||
var fails []string
|
||||
failAt := func(machine, words string) {
|
||||
if v.Machine == "" && machine != "" {
|
||||
v.Machine = machine
|
||||
}
|
||||
fails = append(fails, words)
|
||||
}
|
||||
for _, name := range sortedKeys(c.Unreadable) {
|
||||
failAt("", fmt.Sprintf("the manifest of %s cannot be read: %s", name, c.Unreadable[name]))
|
||||
}
|
||||
for _, name := range sortedKeys(c.Twice) {
|
||||
failAt("", fmt.Sprintf("%s is defined twice, in %s", name, strings.Join(c.Twice[name], " and ")))
|
||||
}
|
||||
machines := Touched(m, c)
|
||||
v.Machines = machines
|
||||
for _, node := range machines {
|
||||
for _, name := range c.Removed {
|
||||
if slices.Contains(m.Assigned[node], name) {
|
||||
failAt(node, fmt.Sprintf("%s runs %s, which the change removes from its source", node, name))
|
||||
}
|
||||
}
|
||||
}
|
||||
trunk := input(m.Manifests, m.Assigned)
|
||||
changed := map[string]catalogue.Manifest{}
|
||||
for name, man := range m.Manifests {
|
||||
changed[name] = man
|
||||
}
|
||||
for name, man := range c.Manifests {
|
||||
changed[name] = man
|
||||
}
|
||||
for _, name := range c.Removed {
|
||||
delete(changed, name)
|
||||
}
|
||||
after := input(changed, m.Assigned)
|
||||
for _, node := range machines {
|
||||
_, before := compose.Resolve(trunk, node)
|
||||
_, now := compose.Resolve(after, node)
|
||||
switch {
|
||||
case now == nil:
|
||||
case before != nil:
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("%s does not compose on the trunk either, so the change is not blamed: %v", node, before))
|
||||
default:
|
||||
failAt(node, fmt.Sprintf("%s composes on the trunk and not with the change: %v", node, now))
|
||||
}
|
||||
}
|
||||
if c.Rebuilt > delivery.CompositionRebuildWarning {
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("the merge rebuilds %d modules, more than %d", c.Rebuilt, delivery.CompositionRebuildWarning))
|
||||
}
|
||||
if c.SDKHead != "" {
|
||||
v.Warnings = append(v.Warnings, fmt.Sprintf("composed with the assigner's own mesh-sdk %s, not the group's mesh-sdk head %s, "+
|
||||
"and a field the running controller cannot read is not judged (issues 1621, 1622, 1623)", sdk, c.SDKHead))
|
||||
}
|
||||
v.Warnings = append(v.Warnings, NotJudged)
|
||||
if len(machines) == 0 {
|
||||
v.Warnings = append(v.Warnings, "the change touches no machine: no module it moves is assigned anywhere")
|
||||
}
|
||||
if len(fails) > 0 {
|
||||
v.State, v.Words = delivery.CompositionFail, strings.Join(fails, "; ")
|
||||
return v
|
||||
}
|
||||
v.State = delivery.CompositionPass
|
||||
return v
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,116 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func meshOf() Mesh {
|
||||
return Mesh{
|
||||
Manifests: map[string]catalogue.Manifest{
|
||||
"mail": {Module: "mail", Version: "1", Requires: []string{"database"}},
|
||||
"store": {Module: "store", Version: "1", Provides: []catalogue.Offer{{Name: "database"}}},
|
||||
"web": {Module: "web", Version: "1"},
|
||||
},
|
||||
Assigned: map[string][]string{"ace": {"mail", "store"}, "g14": {"web"}},
|
||||
}
|
||||
}
|
||||
|
||||
func hasWarning(v delivery.CompositionVerdict, part string) bool {
|
||||
return slices.ContainsFunc(v.Warnings, func(w string) bool { return strings.Contains(w, part) })
|
||||
}
|
||||
|
||||
// 1297/AC1 (its unit half): a change touching one module on one machine composes that machine alone, passes, names
|
||||
// the machine and the mesh-sdk it composed with, and says what it did not judge.
|
||||
func TestAChangeOfOneModuleComposesItsMachine(t *testing.T) {
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"web": {Module: "web", Version: "2"}}}
|
||||
v := Compose(meshOf(), c, "v0.1.14-x", 0)
|
||||
if v.State != delivery.CompositionPass || !slices.Equal(v.Machines, []string{"g14"}) || v.SDK != "v0.1.14-x" {
|
||||
t.Fatalf("composition = %+v, want a pass on g14 alone", v)
|
||||
}
|
||||
if !hasWarning(v, "issue 1461") {
|
||||
t.Errorf("the verdict does not say what it did not judge: %v", v.Warnings)
|
||||
}
|
||||
if _, err := delivery.EncodeCompositionVerdict(v); err != nil {
|
||||
t.Errorf("the verdict cannot be written: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/AC2: a change that removes a module a machine runs fails, naming the machine and the module.
|
||||
func TestRemovingAModuleAMachineRunsFails(t *testing.T) {
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Removed: []string{"web"}}
|
||||
v := Compose(meshOf(), c, "sdk", 0)
|
||||
if v.State != delivery.CompositionFail || v.Machine != "g14" || !strings.Contains(v.Words, "g14 runs web") {
|
||||
t.Fatalf("composition = %+v, want a fail naming g14 and web", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3c: a machine that composes on the trunk and not with the change fails it, naming the machine and why; the
|
||||
// consumer's machine is touched when its provider changes.
|
||||
func TestAMachineTheChangeBreaksFails(t *testing.T) {
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"store": {Module: "store", Version: "2"}}}
|
||||
v := Compose(meshOf(), c, "sdk", 0)
|
||||
if v.State != delivery.CompositionFail || v.Machine != "ace" || !strings.Contains(v.Words, "database") {
|
||||
t.Fatalf("composition = %+v, want a fail on ace naming the provision nothing provides", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3c: what was already broken on the trunk is said and fails nothing.
|
||||
func TestWhatWasBrokenBeforeFailsNothing(t *testing.T) {
|
||||
m := meshOf()
|
||||
m.Manifests["store"] = catalogue.Manifest{Module: "store", Version: "1"}
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"mail": {Module: "mail", Version: "2", Requires: []string{"database"}}}}
|
||||
v := Compose(m, c, "sdk", 0)
|
||||
if v.State != delivery.CompositionPass || !hasWarning(v, "on the trunk either") {
|
||||
t.Fatalf("composition = %+v, want a pass saying ace was already broken", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3c: an unreadable manifest and a module defined twice fail; more than twelve rebuilt modules warn.
|
||||
func TestUnreadableTwiceAndWide(t *testing.T) {
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Unreadable: map[string]string{"web": "unknown field \"scope\""}, Rebuilt: 13}
|
||||
v := Compose(meshOf(), c, "sdk", 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "web cannot be read") || !hasWarning(v, "rebuilds 13 modules") {
|
||||
t.Fatalf("composition = %+v", v)
|
||||
}
|
||||
c = Change{Group: "g", Heads: "0123456789ab", Twice: map[string][]string{"web": {"novox/a", "novox/b"}}, Rebuilt: 12}
|
||||
v = Compose(meshOf(), c, "sdk", 0)
|
||||
if v.State != delivery.CompositionFail || !strings.Contains(v.Words, "web is defined twice") || hasWarning(v, "rebuilds") {
|
||||
t.Fatalf("composition = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1622's criterion as R4g delivers it: a group with a mesh-sdk member is composed with the assigner's own mesh-sdk, and
|
||||
// the verdict says so, naming the member's head it did not use; a group without one says nothing of it.
|
||||
func TestAGroupWithAnSDKMemberSaysWhichSDKComposedIt(t *testing.T) {
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"web": {Module: "web", Version: "2"}}, SDKHead: "f625d98713e4"}
|
||||
v := Compose(meshOf(), c, "v0.1.14-own", 0)
|
||||
if !hasWarning(v, "not the group's mesh-sdk head f625d98713e4") || !hasWarning(v, "v0.1.14-own") || !hasWarning(v, "1623") {
|
||||
t.Errorf("the verdict does not say it composed with its own mesh-sdk: %v", v.Warnings)
|
||||
}
|
||||
c.SDKHead = ""
|
||||
if v := Compose(meshOf(), c, "v0.1.14-own", 0); hasWarning(v, "mesh-sdk head") {
|
||||
t.Errorf("a group without a mesh-sdk member says it has one: %v", v.Warnings)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3a: the machines touched are those running a moved module or a consumer of what it provides; an added module
|
||||
// assigned nowhere touches none, and the verdict says so.
|
||||
func TestTouched(t *testing.T) {
|
||||
m := meshOf()
|
||||
if got := Touched(m, Change{Manifests: map[string]catalogue.Manifest{"store": m.Manifests["store"]}}); !slices.Equal(got, []string{"ace"}) {
|
||||
t.Errorf("touched by store = %v", got)
|
||||
}
|
||||
m.Assigned["g14"] = append(m.Assigned["g14"], "mail")
|
||||
if got := Touched(m, Change{Manifests: map[string]catalogue.Manifest{"store": m.Manifests["store"]}}); !slices.Equal(got, []string{"ace", "g14"}) {
|
||||
t.Errorf("touched by store with a consumer on g14 = %v", got)
|
||||
}
|
||||
c := Change{Group: "g", Heads: "0123456789ab", Manifests: map[string]catalogue.Manifest{"new": {Module: "new", Version: "1"}}}
|
||||
if v := Compose(m, c, "sdk", 0); len(v.Machines) != 0 || !hasWarning(v, "touches no machine") || v.State != delivery.CompositionPass {
|
||||
t.Errorf("an added module assigned nowhere: %+v", v)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,205 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// Module is what the six rules read of one module (ADR 0429 §6, ADR 0528 clause 3): its newest fingerprint F at the
|
||||
// trunk (the indexer's modules), whether the builder's build of F is done, whether F is held, whether the controller
|
||||
// issued its credential for F, its manifest at F, and the nodes it is assigned to.
|
||||
type Module struct {
|
||||
Name string
|
||||
FP string
|
||||
Built bool
|
||||
Held bool
|
||||
Granted bool
|
||||
Manifest catalogue.Manifest
|
||||
Nodes []string
|
||||
}
|
||||
|
||||
// Node is what the rules read of the nodes: whether each is online (the watchman's presence) and each module's have.
|
||||
type Nodes struct {
|
||||
Online map[string]bool
|
||||
Haves map[string]map[string]delivery.Have
|
||||
// Providers is, by provision, the modules that provide it, with the nodes each is assigned to.
|
||||
Providers map[string]map[string][]string
|
||||
}
|
||||
|
||||
// healthy says whether a have runs fp healthy.
|
||||
func healthy(h delivery.Have, fp string) bool {
|
||||
return h.FP == fp && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK
|
||||
}
|
||||
|
||||
// Canary is the module's canary: the first of its nodes, by name, that is online (ADR 0435 §3; one ring, ordered by
|
||||
// name, ADR 0528 clause 3). Empty when none is online.
|
||||
func Canary(nodes []string, online map[string]bool) string {
|
||||
sorted := slices.Sorted(slices.Values(nodes))
|
||||
for _, n := range sorted {
|
||||
if online[n] {
|
||||
return n
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// Holds is whether every rule holds for moving node's want for m to m.FP, and if not, the first rule that does not,
|
||||
// with why (ADR 0429 §6). Rule 4, at most `step` others ahead, always holds: the mesh keeps no step, and step 0 is no
|
||||
// limit (ADR 0528 clause 3).
|
||||
func Holds(m Module, node string, n Nodes) (bool, int, string) {
|
||||
switch {
|
||||
case !m.Built:
|
||||
return false, delivery.RuleBuildDone, fmt.Sprintf("the build of %s at %s is not done", m.Name, short(m.FP))
|
||||
case m.Held:
|
||||
return false, delivery.RuleNotHeld, fmt.Sprintf("%s is held at %s", m.Name, short(m.FP))
|
||||
}
|
||||
canary := Canary(m.Nodes, n.Online)
|
||||
switch {
|
||||
case canary == "":
|
||||
return false, delivery.RuleCanary, fmt.Sprintf("none of the nodes of %s is online, so it has no canary", m.Name)
|
||||
case node != canary && !healthy(n.Haves[canary][m.Name], m.FP):
|
||||
return false, delivery.RuleCanary, fmt.Sprintf("the canary %s does not run %s healthy yet", canary, short(m.FP))
|
||||
}
|
||||
for _, provision := range m.Manifest.Requires {
|
||||
if ok, why := providedHealthy(provision, node, n); !ok {
|
||||
return false, delivery.RuleProviders, why
|
||||
}
|
||||
}
|
||||
if !m.Granted {
|
||||
return false, delivery.RuleGrant, fmt.Sprintf("the controller has not issued the credential of %s for %s", m.Name, short(m.FP))
|
||||
}
|
||||
return true, 0, ""
|
||||
}
|
||||
|
||||
// providedHealthy is rule 5 for one provision: some module providing it runs healthy on the node, or, assigned
|
||||
// elsewhere only, at its canary. A provision no module provides is not a module's to judge (the mesh or the node
|
||||
// offers it) and holds.
|
||||
func providedHealthy(provision, node string, n Nodes) (bool, string) {
|
||||
providers := n.Providers[provision]
|
||||
if len(providers) == 0 {
|
||||
return true, ""
|
||||
}
|
||||
names := make([]string, 0, len(providers))
|
||||
for p := range providers {
|
||||
names = append(names, p)
|
||||
}
|
||||
sort.Strings(names)
|
||||
for _, p := range names {
|
||||
nodes := providers[p]
|
||||
at := node
|
||||
if !slices.Contains(nodes, node) {
|
||||
at = Canary(nodes, n.Online)
|
||||
}
|
||||
if h, ok := n.Haves[at][p]; ok && at != "" && h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK {
|
||||
return true, ""
|
||||
}
|
||||
}
|
||||
return false, fmt.Sprintf("no provider of %s runs healthy on %s or at its canary", provision, node)
|
||||
}
|
||||
|
||||
// keepsData says whether a module holds data a set-back would lose: an own item of a class that keeps (irreplaceable,
|
||||
// valuable or rebuildable; cache and none do not).
|
||||
func keepsData(m catalogue.Manifest) bool {
|
||||
for _, it := range m.DataItems() {
|
||||
if it.Class == catalogue.ClassIrreplaceable || it.Class == catalogue.ClassValuable || it.Class == catalogue.ClassRebuildable {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// Outcome is what one module's pass decides: each node's want to write (only those that change), and a hold to write.
|
||||
type Outcome struct {
|
||||
Wants map[string]delivery.Want
|
||||
// Held is the hold of m.FP to write, when a node's health check failed on it and it is not held yet.
|
||||
Held *delivery.Held
|
||||
// Rules is, by node, the rule holding the node back from m.FP and why, for the comparison.
|
||||
Rules map[string]Holding
|
||||
}
|
||||
|
||||
// Holding is a rule holding a node back, and why.
|
||||
type Holding struct {
|
||||
Rule int
|
||||
Words string
|
||||
}
|
||||
|
||||
// Wants is the pass of one module m: for each of its nodes, the want by the six rules, starting from the want written
|
||||
// before (wants, by node); lastHealthy is, by node, the newest fingerprint other than m.FP the node ran healthy. A failed
|
||||
// health check of m.FP on a node holds it: every node wanting m.FP is set back to its last healthy fingerprint, but a
|
||||
// module holding data is not set back (in release 5 it is asked about from release 6 on: wants are only compared). A
|
||||
// node that wants m but is no longer among its nodes gets its want written removed, never deleted (ADR 0408 §4).
|
||||
func Wants(m Module, n Nodes, wants map[string]delivery.Want, lastHealthy map[string]string) Outcome {
|
||||
out := Outcome{Wants: map[string]delivery.Want{}, Rules: map[string]Holding{}}
|
||||
held := m.Held
|
||||
if !held && m.FP != "" {
|
||||
for _, node := range slices.Sorted(slices.Values(m.Nodes)) {
|
||||
if h, ok := n.Haves[node][m.Name]; ok && h.FP == m.FP && h.Health == delivery.HealthFailed {
|
||||
out.Held = &delivery.Held{Node: node, Words: h.Detail}
|
||||
held = true
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
mm := m
|
||||
mm.Held = held
|
||||
for _, node := range m.Nodes {
|
||||
if m.FP == "" {
|
||||
break
|
||||
}
|
||||
before, had := wants[node]
|
||||
var next delivery.Want
|
||||
switch ok, rule, words := Holds(mm, node, n); {
|
||||
case ok:
|
||||
next = delivery.Want{FP: m.FP}
|
||||
case held && had && before.FP == m.FP:
|
||||
out.Rules[node] = Holding{Rule: rule, Words: words}
|
||||
last := lastHealthy[node]
|
||||
switch {
|
||||
case keepsData(m.Manifest):
|
||||
next = before
|
||||
next.Words = fmt.Sprintf("%s holds data, so it is not set back from %s: the operator is asked from release 6", m.Name, short(m.FP))
|
||||
case last == "":
|
||||
next = before
|
||||
next.Words = fmt.Sprintf("%s is held at %s and %s ran no other fingerprint healthy to set it back to", m.Name, short(m.FP), node)
|
||||
default:
|
||||
next = delivery.Want{FP: last, Newest: m.FP, Rule: rule, Words: words}
|
||||
}
|
||||
default:
|
||||
out.Rules[node] = Holding{Rule: rule, Words: words}
|
||||
if !had || before.Removed {
|
||||
continue
|
||||
}
|
||||
next = delivery.Want{FP: before.FP, Newest: m.FP, Rule: rule, Words: words}
|
||||
if before.FP == m.FP {
|
||||
next = delivery.Want{FP: m.FP}
|
||||
}
|
||||
}
|
||||
if !had || !delivery.SameWant(normal(before), normal(next)) {
|
||||
out.Wants[node] = next
|
||||
}
|
||||
}
|
||||
for node, before := range wants {
|
||||
if !slices.Contains(m.Nodes, node) && !before.Removed {
|
||||
out.Wants[node] = delivery.Want{Removed: true, Words: fmt.Sprintf("%s is no longer assigned to %s", m.Name, node)}
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// normal is a want without its version, as a decoded and a composed one are compared.
|
||||
func normal(w delivery.Want) delivery.Want {
|
||||
w.V = 0
|
||||
return w
|
||||
}
|
||||
|
||||
func short(fp string) string {
|
||||
part, err := delivery.FingerprintKeyPart(fp)
|
||||
if err != nil || len(part) < 12 {
|
||||
return fp
|
||||
}
|
||||
return part[:12]
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
package assign
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func fpOf(c byte) string { return delivery.SourcePrefix2 + strings.Repeat(string(c), 64) }
|
||||
|
||||
var (
|
||||
fNew = fpOf('b')
|
||||
fOld = fpOf('a')
|
||||
)
|
||||
|
||||
// ready is a module mail at fNew on ace and g14 for which every rule holds: built, not held, granted, its provider
|
||||
// store healthy on ace and the canary ace running fNew healthy.
|
||||
func ready() (Module, Nodes) {
|
||||
m := Module{Name: "mail", FP: fNew, Built: true, Granted: true, Nodes: []string{"g14", "ace"},
|
||||
Manifest: catalogue.Manifest{Module: "mail", Requires: []string{"database"}}}
|
||||
n := Nodes{
|
||||
Online: map[string]bool{"ace": true, "g14": true},
|
||||
Haves: map[string]map[string]delivery.Have{
|
||||
"ace": {"mail": {FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}, "store": {FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK}},
|
||||
"g14": {"mail": {FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}},
|
||||
},
|
||||
Providers: map[string]map[string][]string{"database": {"store": {"ace"}}},
|
||||
}
|
||||
return m, n
|
||||
}
|
||||
|
||||
func wantedOld() map[string]delivery.Want {
|
||||
return map[string]delivery.Want{"ace": {FP: fOld}, "g14": {FP: fOld}}
|
||||
}
|
||||
|
||||
// 1297/B4: with every rule holding, both nodes' wants move to the new fingerprint (the canary ace runs it healthy, so
|
||||
// g14 follows at once: step 0, ADR 0528 clause 3).
|
||||
func TestEveryRuleHoldsSoTheWantsMove(t *testing.T) {
|
||||
m, n := ready()
|
||||
out := Wants(m, n, wantedOld(), nil)
|
||||
for _, node := range []string{"ace", "g14"} {
|
||||
if out.Wants[node].FP != fNew || out.Wants[node].Rule != 0 {
|
||||
t.Errorf("%s's want = %+v, want %s", node, out.Wants[node], fNew)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B4, one test per rule: given every rule but N holds, no want moves to F, and the rule N is named.
|
||||
func TestEachRuleHoldsTheWantBack(t *testing.T) {
|
||||
for rule, breakIt := range map[int]func(*Module, *Nodes){
|
||||
delivery.RuleBuildDone: func(m *Module, _ *Nodes) { m.Built = false },
|
||||
delivery.RuleNotHeld: func(m *Module, _ *Nodes) { m.Held = true },
|
||||
delivery.RuleCanary: func(_ *Module, n *Nodes) {
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
|
||||
},
|
||||
delivery.RuleProviders: func(_ *Module, n *Nodes) {
|
||||
n.Haves["ace"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthWaits}
|
||||
},
|
||||
delivery.RuleGrant: func(m *Module, _ *Nodes) { m.Granted = false },
|
||||
} {
|
||||
m, n := ready()
|
||||
breakIt(&m, &n)
|
||||
out := Wants(m, n, wantedOld(), nil)
|
||||
if w, ok := out.Wants["g14"]; ok && w.FP == fNew {
|
||||
t.Errorf("rule %d broken: g14's want moved to F", rule)
|
||||
}
|
||||
if got := out.Rules["g14"].Rule; got != rule {
|
||||
t.Errorf("rule %d broken: g14 is held back by rule %d (%s)", rule, got, out.Rules["g14"].Words)
|
||||
}
|
||||
if w := out.Wants["g14"]; w.FP != "" && (w.FP != fOld || w.Newest != fNew || w.Rule != rule) {
|
||||
t.Errorf("rule %d broken: g14's want is %+v", rule, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B4, ADR 0435 §3: a canary that is offline is passed over: the next node by name is the canary, and moves first.
|
||||
func TestAnOfflineCanaryIsPassedOver(t *testing.T) {
|
||||
m, n := ready()
|
||||
n.Online["ace"] = false
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fOld, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
|
||||
n.Providers["database"]["store"] = []string{"ace", "g14"}
|
||||
n.Haves["g14"]["store"] = delivery.Have{FP: fpOf('s'), Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
|
||||
if c := Canary(m.Nodes, n.Online); c != "g14" {
|
||||
t.Fatalf("canary = %q, want g14", c)
|
||||
}
|
||||
out := Wants(m, n, wantedOld(), nil)
|
||||
if out.Wants["g14"].FP != fNew {
|
||||
t.Errorf("the canary g14's want = %+v", out.Wants["g14"])
|
||||
}
|
||||
if out.Rules["ace"].Rule != delivery.RuleCanary {
|
||||
t.Errorf("ace, behind the canary g14, is held back by %+v", out.Rules["ace"])
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B4: a failed health check of F holds F and sets the nodes that want it back to their last healthy fingerprint.
|
||||
func TestAFailedHealthHoldsAndSetsBack(t *testing.T) {
|
||||
m, n := ready()
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed, Detail: "connection refused"}
|
||||
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}, "g14": {FP: fOld}}, map[string]string{"ace": fOld})
|
||||
if out.Held == nil || out.Held.Node != "ace" || out.Held.Words != "connection refused" {
|
||||
t.Fatalf("held = %+v", out.Held)
|
||||
}
|
||||
if w := out.Wants["ace"]; w.FP != fOld || w.Newest != fNew || w.Rule != delivery.RuleNotHeld {
|
||||
t.Errorf("ace's want = %+v, want set back to the last healthy fingerprint", w)
|
||||
}
|
||||
if w, ok := out.Wants["g14"]; ok && w.FP == fNew {
|
||||
t.Errorf("g14 moved to a held fingerprint: %+v", w)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B4: a module holding data is not set back; its want stays and says why.
|
||||
func TestADataModuleIsNotSetBack(t *testing.T) {
|
||||
m, n := ready()
|
||||
m.Manifest.Data = &catalogue.Data{Own: []catalogue.DataItem{{ID: "mail", Class: catalogue.ClassIrreplaceable}}}
|
||||
n.Haves["ace"]["mail"] = delivery.Have{FP: fNew, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed}
|
||||
out := Wants(m, n, map[string]delivery.Want{"ace": {FP: fNew}}, map[string]string{"ace": fOld})
|
||||
if out.Held == nil {
|
||||
t.Fatal("F was not held")
|
||||
}
|
||||
if w := out.Wants["ace"]; w.FP != fNew || !strings.Contains(w.Words, "holds data") {
|
||||
t.Errorf("ace's want = %+v, want it left at F, saying it holds data", w)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B4, ADR 0408 §4: a node that leaves the module's nodes gets its want written removed, never deleted; a want that
|
||||
// is already as the rules give it is not written again.
|
||||
func TestALeavingNodeIsRemovedAndNothingIsWrittenTwice(t *testing.T) {
|
||||
m, n := ready()
|
||||
m.Nodes = []string{"ace"}
|
||||
out := Wants(m, n, map[string]delivery.Want{"ace": {V: 1, FP: fNew}, "g14": {FP: fOld}}, nil)
|
||||
if w := out.Wants["g14"]; !w.Removed {
|
||||
t.Errorf("g14's want = %+v, want removed", w)
|
||||
}
|
||||
if _, ok := out.Wants["ace"]; ok {
|
||||
t.Errorf("ace's want, unchanged, was written again: %+v", out.Wants["ace"])
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
package manifest
|
||||
|
||||
import (
|
||||
"os"
|
||||
"reflect"
|
||||
"slices"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
func load(t *testing.T) catalogue.Manifest {
|
||||
t.Helper()
|
||||
raw, err := os.ReadFile("module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
m, err := catalogue.ParseManifest(raw)
|
||||
if err != nil {
|
||||
t.Fatalf("mesh-sdk's manifest reader refuses the manifest: %v", err)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// 1297/B2a: the controller's reader (mesh-sdk catalogue, which the controller reads manifests with) takes the
|
||||
// manifest; it reads exactly the buckets the assigner opens (the program's list, held equal in its own test), keeps
|
||||
// its three states, and holds a credential of its own.
|
||||
func TestManifest(t *testing.T) {
|
||||
m := load(t)
|
||||
if m.Module != "mesh-delivery-assigner" {
|
||||
t.Errorf("module = %q", m.Module)
|
||||
}
|
||||
reads := slices.Clone(m.Reads)
|
||||
slices.Sort(reads)
|
||||
want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}
|
||||
if !reflect.DeepEqual(reads, want) {
|
||||
t.Errorf("reads = %v, want exactly %v", reads, want)
|
||||
}
|
||||
states := map[string]int{}
|
||||
for _, s := range m.State {
|
||||
states[s.Name] = s.History
|
||||
}
|
||||
if want := map[string]int{"compositions": 5, "differences": 64, "wants": 20}; !reflect.DeepEqual(states, want) {
|
||||
t.Errorf("state = %v, want %v", states, want)
|
||||
}
|
||||
if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
|
||||
t.Errorf("own-secrets = %v, want exactly broker (or no credential is issued)", m.OwnSecrets)
|
||||
}
|
||||
if m.SecretsOwner != "mesh-delivery-assigner" {
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
if len(m.Claims) != 0 || len(m.Tools) != 0 || len(m.Invokes) != 0 {
|
||||
t.Errorf("the assigner claims, serves or calls something: claims %v, tools %v, invokes %v", m.Claims, m.Tools, m.Invokes)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
{
|
||||
"module": "mesh-delivery-assigner",
|
||||
"version": "1",
|
||||
"slug": "assigner",
|
||||
"reads": [
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
"mesh-controller.grants",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules",
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "compositions", "history": 5 },
|
||||
{ "name": "differences", "history": 64 },
|
||||
{ "name": "wants", "history": 20 }
|
||||
],
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
"secrets-owner": "mesh-delivery-assigner",
|
||||
"resources": [
|
||||
{
|
||||
"id": "account",
|
||||
"type": "user",
|
||||
"name": "mesh-delivery-assigner",
|
||||
"shell": "/usr/bin/nologin",
|
||||
"home": "/var/lib/mesh-delivery-assigner"
|
||||
},
|
||||
{
|
||||
"id": "state",
|
||||
"type": "directory",
|
||||
"mode": "0700",
|
||||
"owner": "mesh-delivery-assigner",
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "assigner",
|
||||
"type": "process",
|
||||
"name": "mesh-delivery-assigner",
|
||||
"artifact": "program",
|
||||
"run": [
|
||||
"./mesh-delivery-assigner"
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker"
|
||||
},
|
||||
"user": "mesh-delivery-assigner",
|
||||
"health": {
|
||||
"kind": "unit"
|
||||
}
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "program",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/mesh-delivery-assigner",
|
||||
"binary": "mesh-delivery-assigner"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"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
|
||||
}
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
//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.
|
||||
func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) {
|
||||
dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "comp")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { os.RemoveAll(dir) })
|
||||
srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go srv.Start()
|
||||
if !srv.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
t.Cleanup(srv.Shutdown)
|
||||
nc, err := nats.Connect(srv.ClientURL())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, _ := jetstream.New(nc)
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
|
||||
defer cancel()
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -27,11 +27,13 @@ type judgeDeps struct {
|
||||
forge check.Forge
|
||||
announce check.Announcer
|
||||
builds check.BuildStates
|
||||
logf func(format string, args ...any)
|
||||
// compositions is the assigner's bucket of compositions; nil leaves every composition not run.
|
||||
compositions check.CompositionReader
|
||||
logf func(format string, args ...any)
|
||||
}
|
||||
|
||||
func newJudge(d judgeDeps) check.Judge {
|
||||
return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places,
|
||||
return &check.Examiner{Compositions: d.compositions, Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places,
|
||||
Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf}
|
||||
}
|
||||
|
||||
|
||||
@@ -222,6 +222,24 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
default:
|
||||
builds = buildReader{kvStore{kv}}
|
||||
}
|
||||
// The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the
|
||||
// composition was not run, until the checker starts again.
|
||||
var compositions check.CompositionReader
|
||||
switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); {
|
||||
case errors.Is(err, jetstream.ErrBucketNotFound):
|
||||
logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket)
|
||||
case err != nil:
|
||||
return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err)
|
||||
default:
|
||||
cw := compositionWatch{kv: kv}
|
||||
if pkv, err := js.KeyValue(ctx, delivery.PresenceBucket); err == nil {
|
||||
cw.presence = pkv
|
||||
}
|
||||
compositions = cw
|
||||
if err := followCompositions(ctx, kv, ch); err != nil {
|
||||
return fmt.Errorf("the bucket %s cannot be watched: %w", delivery.AssignerCompositionsBucket, err)
|
||||
}
|
||||
}
|
||||
// The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries.
|
||||
tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf}
|
||||
holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger)
|
||||
@@ -238,7 +256,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks,
|
||||
kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]},
|
||||
repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf,
|
||||
announce: announce,
|
||||
announce: announce, compositions: compositions,
|
||||
})
|
||||
ch.Announce = announce
|
||||
tryer.Judge = ch.Judge
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
|
||||
"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"
|
||||
}
|
||||
v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words}
|
||||
}
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// 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): 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, 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 || v.State != delivery.VerdictPass || !notJudged(v) {
|
||||
return
|
||||
}
|
||||
withComposition(&v, comp)
|
||||
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)
|
||||
}
|
||||
@@ -0,0 +1,149 @@
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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,80 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// fakeCompositions answers the assigner's composition of a check from memory, recording the keys asked.
|
||||
type fakeCompositions struct {
|
||||
v delivery.CompositionVerdict
|
||||
err error
|
||||
asked []string
|
||||
}
|
||||
|
||||
func (f *fakeCompositions) Await(_ context.Context, key string) (delivery.CompositionVerdict, error) {
|
||||
f.asked = append(f.asked, key)
|
||||
return f.v, f.err
|
||||
}
|
||||
|
||||
// 1297/AC3 (its unit half): a check whose tests pass takes the assigner's composition of it, by the check's own key,
|
||||
// into its verdict: a pass says it passed with the mesh-sdk and machines it composed, and is no longer "not run".
|
||||
func TestAPassingCheckTakesTheAssignersComposition(t *testing.T) {
|
||||
b := newBench(t)
|
||||
f := &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "v0.1.14-x", Machines: []string{"g14"}, Took: 40}}
|
||||
b.ex.Compositions = f
|
||||
cs := caseOf(t, pullOf(repoA, 7, head1))
|
||||
v := b.judge(t, cs).Verdict
|
||||
if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionPass || v.Composition.SDK != "v0.1.14-x" || v.Took.Composition != 40 {
|
||||
t.Fatalf("verdict = %+v, composition %+v", v, v.Composition)
|
||||
}
|
||||
if len(f.asked) != 1 || f.asked[0] != cs.Key {
|
||||
t.Errorf("asked %v, want the check's key %s", f.asked, cs.Key)
|
||||
}
|
||||
if !compositionMentions(v, "passed") {
|
||||
t.Errorf("the line says %q", compositionWords(v.Composition, v.Took.Composition))
|
||||
}
|
||||
}
|
||||
|
||||
func compositionMentions(v delivery.Verdict, word string) bool {
|
||||
return strings.Contains(compositionWords(v.Composition, v.Took.Composition), word)
|
||||
}
|
||||
|
||||
// 1297/AC2 at the checker: a failed composition fails the check, naming the machine it failed at and its words.
|
||||
func TestAFailedCompositionFailsTheCheck(t *testing.T) {
|
||||
b := newBench(t)
|
||||
b.ex.Compositions = &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionFail, SDK: "x",
|
||||
Machine: "g14", Words: "g14 runs web, which the change removes from its source"}}
|
||||
v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
|
||||
if v.State != delivery.VerdictFail || v.Failed.Machine != "g14" || v.Composition == nil || v.Composition.State != delivery.CompositionFail {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
v.Group, v.Heads = "feat/x", "0123456789ab" // the checker names them when it writes the verdict
|
||||
if _, err := delivery.EncodeVerdict(v); err != nil {
|
||||
t.Errorf("the verdict cannot be written: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1297/B3d: a composition that cannot be read leaves it not run; a try is not composed.
|
||||
func TestACompositionNotHadIsNotRun(t *testing.T) {
|
||||
b := newBench(t)
|
||||
f := &fakeCompositions{err: errors.New("the check's bound passed")}
|
||||
b.ex.Compositions = f
|
||||
v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
|
||||
if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionNotRun {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
b2 := newBench(t)
|
||||
f2 := &fakeCompositions{v: delivery.CompositionVerdict{State: delivery.CompositionFail, Words: "w", SDK: "x"}}
|
||||
b2.ex.Compositions = f2
|
||||
cs := caseOf(t, pullOf(repoA, 7, head1))
|
||||
cs.Try = &TryOptions{}
|
||||
b2.judge(t, cs)
|
||||
if len(f2.asked) != 0 {
|
||||
t.Errorf("a try asked for a composition: %v", f2.asked)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package check
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"path"
|
||||
"slices"
|
||||
@@ -105,8 +106,18 @@ type Examiner struct {
|
||||
}
|
||||
// Trunks are the trunk heads of every repository, for the siblings a repository's tests read.
|
||||
Trunks interface{ Heads() map[string]string }
|
||||
Now func() time.Time
|
||||
Logf func(format string, args ...any)
|
||||
// Compositions answers the assigner's composition check of a check by the check's key (delivery.CompositionKey),
|
||||
// waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil
|
||||
// leaves the composition not run.
|
||||
Compositions CompositionReader
|
||||
Now func() time.Time
|
||||
Logf func(format string, args ...any)
|
||||
}
|
||||
|
||||
// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote
|
||||
// a verdict that does not wait, or ctx ends.
|
||||
type CompositionReader interface {
|
||||
Await(ctx context.Context, key string) (delivery.CompositionVerdict, error)
|
||||
}
|
||||
|
||||
func (j *Examiner) now() time.Time {
|
||||
@@ -230,13 +241,36 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) {
|
||||
return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil
|
||||
}
|
||||
|
||||
// finish is the outcome of the verdict with the state of the builds the candidate's modules carry.
|
||||
// finish is the outcome of the verdict with the state of the builds the candidate's modules carry, and, for a check
|
||||
// whose tests passed, the assigner's composition of it.
|
||||
func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome {
|
||||
out := st.outcome(v)
|
||||
if out.Verdict.State == delivery.VerdictPass {
|
||||
st.composed(ctx, &out.Verdict)
|
||||
}
|
||||
out.Builds = st.j.buildNotes(ctx, st.c)
|
||||
return out
|
||||
}
|
||||
|
||||
// composed reads the assigner's composition of the check into its verdict (ADR 0528 clause 5): a failed composition
|
||||
// fails the check, naming the machine it failed at. A try is not composed (it has no candidate); a composition that
|
||||
// cannot be had leaves it not run, said; with no assigner running it is not judged, said at once.
|
||||
func (st *state) composed(ctx context.Context, v *delivery.Verdict) {
|
||||
if st.j.Compositions == nil || st.c.Try != nil {
|
||||
return
|
||||
}
|
||||
comp, err := st.j.Compositions.Await(ctx, st.c.Key)
|
||||
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)
|
||||
return
|
||||
}
|
||||
withComposition(v, comp)
|
||||
}
|
||||
|
||||
// outcome fills the verdict's levels, times, machines and composition from what ran.
|
||||
func (st *state) outcome(v delivery.Verdict) Outcome {
|
||||
if v.State == delivery.VerdictPass || len(v.Levels) == 0 {
|
||||
|
||||
@@ -99,6 +99,9 @@ func compositionWords(c *delivery.Composition, took int64) string { return compo
|
||||
|
||||
// composition is compositionWords; short counts the machines and leaves the SDK commit to the report.
|
||||
func composition(c *delivery.Composition, took int64, short bool) string {
|
||||
if c != nil && c.State == delivery.CompositionNotRun && c.Words != "" {
|
||||
return "composition not judged: " + c.Words
|
||||
}
|
||||
if c == nil || c.State == delivery.CompositionNotRun {
|
||||
return "composition not yet run"
|
||||
}
|
||||
|
||||
@@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
"mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-controller.plans-building", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
|
||||
@@ -22,6 +22,7 @@
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.plans-building",
|
||||
"mesh-delivery-assigner.compositions",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023134155-3e5d4c30ffa7
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -23,8 +23,9 @@ const (
|
||||
// NonePreviewConflict opens the reason of a module whose pull request's head does not merge into the trunk; the files
|
||||
// in conflict follow.
|
||||
NonePreviewConflict = "does not merge: the head conflicts with the trunk in "
|
||||
// NonePreviewNoModule is for a module whose directory the preview names no manifest for.
|
||||
NonePreviewNoModule = "the merge preview names no manifest for the module's directory"
|
||||
// NonePreviewNoModule is for a module whose directory the preview names no manifest for: mesh-sdk's words, which the
|
||||
// assigner reads as the change removing the module (issue 1297).
|
||||
NonePreviewNoModule = delivery.CandidateNoneNoModule
|
||||
// NonePreviewNoManifest is for a module whose manifest at the merge the indexer does not hold.
|
||||
NonePreviewNoManifest = "the manifest of the merge is not kept"
|
||||
// NoneNoDirectory is for a module whose directory in its repository neither the modules bucket nor the preview tells.
|
||||
|
||||
Reference in New Issue
Block a user