From 034a6de9a706b7f9958597a6d0bb7b7208aaa17a Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:19:06 +0200 Subject: [PATCH 01/10] =?UTF-8?q?Make=20the=20module=20mesh-delivery-assig?= =?UTF-8?q?ner:=20its=20manifest's=20reads=20and=20states,=20its=20own=20c?= =?UTF-8?q?redential=20and=20the=20states=20it=20announces,=20so=20the=20c?= =?UTF-8?q?ontroller=20grants=20it=20what=20R5=20reads=20(issue=201297,=20?= =?UTF-8?q?subtask=20B2a;=20ADR=200528,=20ADR=200429=20=C2=A71)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-assigner/main.go | 127 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/main_test.go | 59 ++++++++ modules/mesh-delivery-assigner/go.mod | 9 ++ modules/mesh-delivery-assigner/go.sum | 27 ++++ .../mesh-delivery-assigner/manifest_test.go | 57 ++++++++ modules/mesh-delivery-assigner/module.json | 70 ++++++++++ 6 files changed, 349 insertions(+) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go create mode 100644 modules/mesh-delivery-assigner/go.mod create mode 100644 modules/mesh-delivery-assigner/go.sum create mode 100644 modules/mesh-delivery-assigner/manifest_test.go create mode 100644 modules/mesh-delivery-assigner/module.json diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go new file mode 100644 index 0000000..4ac7045 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -0,0 +1,127 @@ +// 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..`, 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)) + <-ctx.Done() + return nil +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go new file mode 100644 index 0000000..f52e1fb --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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 +// (`.` read as the bucket `_`), 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) + } +} diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod new file mode 100644 index 0000000..4998342 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.mod @@ -0,0 +1,9 @@ +module mesh-delivery-assigner + +go 1.26.0 + +require ( + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a + github.com/nats-io/nats-server/v2 v2.11.17 + github.com/nats-io/nats.go v1.54.0 +) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum new file mode 100644 index 0000000..5c1a857 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.sum @@ -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= diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go new file mode 100644 index 0000000..417ac68 --- /dev/null +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json new file mode 100644 index 0000000..599c3f6 --- /dev/null +++ b/modules/mesh-delivery-assigner/module.json @@ -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" + } + ] + } +} From c2b7b8867ea61e6b952b25ce8162f9ecadba26d5 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:19:49 +0200 Subject: [PATCH 02/10] Start the assigner whole on an embedded bus in an integration test: it says its three states and refuses to start without a bucket it reads (issue 1297, subtask B2b) --- .../integration_test.go | 144 ++++++++++++++++++ 1 file changed, 144 insertions(+) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go new file mode 100644 index 0000000..739b315 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -0,0 +1,144 @@ +//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/process" +) + +// 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) + } + } +} From d8599855e14d8dd825408a2fdd9c10f9406b61aa Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:25:22 +0200 Subject: [PATCH 03/10] Compose the machines a change touches, from the trunk's and the change's manifests and the controller's assignments on the bus, and write the verdict under the checker's key; what needs the facts snapshot is said as not judged (issue 1297, subtasks B3a-B3c; ADR 0446 clauses 1-3, ADR 0528 clause 5, ADR 0482 clause 3 as R4g delivers it) --- .../integration_test.go | 70 ++++++ .../cmd/mesh-delivery-assigner/main.go | 95 ++++++- .../cmd/mesh-delivery-assigner/pass.go | 81 ++++++ .../cmd/mesh-delivery-assigner/world.go | 231 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/world_test.go | 90 +++++++ .../internal/assign/compose.go | 196 +++++++++++++++ .../internal/assign/compose_test.go | 116 +++++++++ 7 files changed, 877 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/compose.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/compose_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 739b315..d45b86a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -17,7 +17,9 @@ import ( "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 @@ -142,3 +144,71 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { } } } + +// 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) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 4ac7045..9bfae45 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -122,6 +122,97 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro 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)) - <-ctx.Done() - return nil + 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 { + if err := watch(name); 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) + } + } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go new file mode 100644 index 0000000..f5c55fc --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -0,0 +1,81 @@ +package main + +import ( + "context" + "errors" + "time" + + "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) +} + +// 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) +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go new file mode 100644 index 0000000..e96c070 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -0,0 +1,231 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "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-delivery-indexer internal/index/previews.go NonePreviewNoModule): the change removes the module. +const noModuleWords = "the merge preview names no manifest for the module's directory" + +// 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 +} + +// 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{}} +} + +// 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 +} + +// 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, "" +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go new file mode 100644 index 0000000..ee3a8e3 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compose.go b/modules/mesh-delivery-assigner/internal/assign/compose.go new file mode 100644 index 0000000..6de7006 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compose.go @@ -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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compose_test.go b/modules/mesh-delivery-assigner/internal/assign/compose_test.go new file mode 100644 index 0000000..889988a --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compose_test.go @@ -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) + } +} From 5091cab65274460157b8040cc433b13a15c2b0ee Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:27:59 +0200 Subject: [PATCH 04/10] Read the assigner's composition into the checker's verdict in place of 'not yet run', failing a check whose composition failed, naming its machine (issue 1297, subtask B3d; ADR 0528 clause 5) --- .../cmd/mesh-delivery-checker/compositions.go | 45 ++++++++++ .../compositions_integration_test.go | 82 +++++++++++++++++++ .../cmd/mesh-delivery-checker/judge.go | 6 +- .../cmd/mesh-delivery-checker/main.go | 13 ++- modules/mesh-delivery-checker/go.mod | 2 +- .../internal/check/composition_test.go | 80 ++++++++++++++++++ .../internal/check/judge.go | 43 +++++++++- .../mesh-delivery-checker/manifest_test.go | 2 +- modules/mesh-delivery-checker/module.json | 1 + 9 files changed, 266 insertions(+), 8 deletions(-) create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go create mode 100644 modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go create mode 100644 modules/mesh-delivery-checker/internal/check/composition_test.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go new file mode 100644 index 0000000..951b0a6 --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -0,0 +1,45 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// compositionWatch reads the assigner's composition of a check (ADR 0528 clause 5) by watching its key: no poll and no +// timer, the check's own bound ends the wait. +type compositionWatch struct{ kv jetstream.KeyValue } + +// Await answers the composition under key once the assigner wrote one that does not wait; a value it cannot read is an +// error, and ctx ending is ctx's error. +func (c compositionWatch) Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) { + w, err := c.kv.Watch(ctx, key) + if err != nil { + return delivery.CompositionVerdict{}, err + } + defer w.Stop() + for { + select { + case <-ctx.Done(): + return delivery.CompositionVerdict{}, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + v, err := delivery.DecodeCompositionVerdict(e.Value()) + if err != nil { + return delivery.CompositionVerdict{}, fmt.Errorf("the composition %s cannot be read: %w", key, err) + } + if v.State != delivery.CompositionWaits { + return v, nil + } + } + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go new file mode 100644 index 0000000..070abae --- /dev/null +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -0,0 +1,82 @@ +//go:build integration + +package main + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/nats-io/nats-server/v2/server" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// 1297/AC3 (its bus half): on the real bus, the checker's read of the assigner's composition waits while the key is +// absent and while the assigner says it waits, and answers the verdict the assigner then writes under the check's key. +func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { + dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "comp") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { os.RemoveAll(dir) }) + srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true}) + if err != nil { + t.Fatal(err) + } + go srv.Start() + if !srv.ReadyForConnections(10 * time.Second) { + t.Fatal("the bus did not come up") + } + t.Cleanup(srv.Shutdown) + nc, err := nats.Connect(srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.AssignerCompositionsBucket, History: delivery.AssignerCompositionsHistory}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.CompositionKey("feat/x", "0123456789ab") + got := make(chan delivery.CompositionVerdict, 1) + go func() { + v, err := compositionWatch{kv}.Await(ctx, key) + if err != nil { + t.Error(err) + } + got <- v + }() + put := func(v delivery.CompositionVerdict) { + data, err := delivery.EncodeCompositionVerdict(v) + if err != nil { + t.Fatal(err) + } + if _, err := kv.Put(ctx, key, data); err != nil { + t.Fatal(err) + } + } + time.Sleep(200 * time.Millisecond) + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionWaits, Words: "the indexer holds no candidates yet"}) + select { + case v := <-got: + t.Fatalf("a waiting composition was answered: %+v", v) + case <-time.After(300 * time.Millisecond): + } + put(delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "x", Machines: []string{"g14"}}) + select { + case v := <-got: + if v.State != delivery.CompositionPass || v.Machines[0] != "g14" { + t.Errorf("answered %+v", v) + } + case <-time.After(5 * time.Second): + t.Fatal("the composition written was not answered") + } +} diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go index ed8ff96..ecc4e6b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/judge.go @@ -27,11 +27,13 @@ type judgeDeps struct { forge check.Forge announce check.Announcer builds check.BuildStates - logf func(format string, args ...any) + // compositions is the assigner's bucket of compositions; nil leaves every composition not run. + compositions check.CompositionReader + logf func(format string, args ...any) } func newJudge(d judgeDeps) check.Judge { - return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, + return &check.Examiner{Compositions: d.compositions, Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 41377e8..676d380 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -222,6 +222,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro default: builds = buildReader{kvStore{kv}} } + // The assigner's compositions (ADR 0528 clause 5, issue 1297); where its bucket is not there yet a verdict says the + // composition was not run, until the checker starts again. + var compositions check.CompositionReader + switch kv, err := js.KeyValue(ctx, delivery.AssignerCompositionsBucket); { + case errors.Is(err, jetstream.ErrBucketNotFound): + logger.Printf("the bucket %s does not exist yet: a verdict says the composition was not run until the checker starts again", delivery.AssignerCompositionsBucket) + case err != nil: + return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) + default: + compositions = compositionWatch{kv} + } // The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries. tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf} holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger) @@ -238,7 +249,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro runner: runner, tools: tools, presence: presence, have: have, places: places, trunks: trunks, kept: kvStore{buckets[delivery.CheckerKeptBucket]}, previews: kvStore{buckets[delivery.PreviewsBucket]}, repos: repos, builds: builds, forge: forgeReads{call: caller.Call}, logf: logger.Printf, - announce: announce, + announce: announce, compositions: compositions, }) ch.Announce = announce tryer.Judge = ch.Judge diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 0de4a55..e42ae23 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/internal/check/composition_test.go b/modules/mesh-delivery-checker/internal/check/composition_test.go new file mode 100644 index 0000000..19bb6b6 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/composition_test.go @@ -0,0 +1,80 @@ +package check + +import ( + "context" + "errors" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// fakeCompositions answers the assigner's composition of a check from memory, recording the keys asked. +type fakeCompositions struct { + v delivery.CompositionVerdict + err error + asked []string +} + +func (f *fakeCompositions) Await(_ context.Context, key string) (delivery.CompositionVerdict, error) { + f.asked = append(f.asked, key) + return f.v, f.err +} + +// 1297/AC3 (its unit half): a check whose tests pass takes the assigner's composition of it, by the check's own key, +// into its verdict: a pass says it passed with the mesh-sdk and machines it composed, and is no longer "not run". +func TestAPassingCheckTakesTheAssignersComposition(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionPass, SDK: "v0.1.14-x", Machines: []string{"g14"}, Took: 40}} + b.ex.Compositions = f + cs := caseOf(t, pullOf(repoA, 7, head1)) + v := b.judge(t, cs).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionPass || v.Composition.SDK != "v0.1.14-x" || v.Took.Composition != 40 { + t.Fatalf("verdict = %+v, composition %+v", v, v.Composition) + } + if len(f.asked) != 1 || f.asked[0] != cs.Key { + t.Errorf("asked %v, want the check's key %s", f.asked, cs.Key) + } + if !compositionMentions(v, "passed") { + t.Errorf("the line says %q", compositionWords(v.Composition, v.Took.Composition)) + } +} + +func compositionMentions(v delivery.Verdict, word string) bool { + return strings.Contains(compositionWords(v.Composition, v.Took.Composition), word) +} + +// 1297/AC2 at the checker: a failed composition fails the check, naming the machine it failed at and its words. +func TestAFailedCompositionFailsTheCheck(t *testing.T) { + b := newBench(t) + b.ex.Compositions = &fakeCompositions{v: delivery.CompositionVerdict{Group: "feat/x", Heads: "0123456789ab", State: delivery.CompositionFail, SDK: "x", + Machine: "g14", Words: "g14 runs web, which the change removes from its source"}} + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictFail || v.Failed.Machine != "g14" || v.Composition == nil || v.Composition.State != delivery.CompositionFail { + t.Fatalf("verdict = %+v", v) + } + v.Group, v.Heads = "feat/x", "0123456789ab" // the checker names them when it writes the verdict + if _, err := delivery.EncodeVerdict(v); err != nil { + t.Errorf("the verdict cannot be written: %v", err) + } +} + +// 1297/B3d: a composition that cannot be read leaves it not run; a try is not composed. +func TestACompositionNotHadIsNotRun(t *testing.T) { + b := newBench(t) + f := &fakeCompositions{err: errors.New("the check's bound passed")} + b.ex.Compositions = f + v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict + if v.State != delivery.VerdictPass || v.Composition == nil || v.Composition.State != delivery.CompositionNotRun { + t.Fatalf("verdict = %+v", v) + } + b2 := newBench(t) + f2 := &fakeCompositions{v: delivery.CompositionVerdict{State: delivery.CompositionFail, Words: "w", SDK: "x"}} + b2.ex.Compositions = f2 + cs := caseOf(t, pullOf(repoA, 7, head1)) + cs.Try = &TryOptions{} + b2.judge(t, cs) + if len(f2.asked) != 0 { + t.Errorf("a try asked for a composition: %v", f2.asked) + } +} diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index dcd07f0..099caca 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -105,8 +105,18 @@ type Examiner struct { } // Trunks are the trunk heads of every repository, for the siblings a repository's tests read. Trunks interface{ Heads() map[string]string } - Now func() time.Time - Logf func(format string, args ...any) + // Compositions answers the assigner's composition check of a check by the check's key (delivery.CompositionKey), + // waiting until the assigner wrote a verdict that does not wait, or ctx ends (ADR 0528 clause 5, issue 1297); nil + // leaves the composition not run. + Compositions CompositionReader + Now func() time.Time + Logf func(format string, args ...any) +} + +// CompositionReader answers the assigner's composition of a check by the check's key, waiting until the assigner wrote +// a verdict that does not wait, or ctx ends. +type CompositionReader interface { + Await(ctx context.Context, key string) (delivery.CompositionVerdict, error) } func (j *Examiner) now() time.Time { @@ -230,13 +240,40 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) { return st.finish(ctx, delivery.Verdict{State: delivery.VerdictPass}), nil } -// finish is the outcome of the verdict with the state of the builds the candidate's modules carry. +// finish is the outcome of the verdict with the state of the builds the candidate's modules carry, and, for a check +// whose tests passed, the assigner's composition of it. func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { out := st.outcome(v) + if out.Verdict.State == delivery.VerdictPass { + st.composed(ctx, &out.Verdict) + } out.Builds = st.j.buildNotes(ctx, st.c) return out } +// composed reads the assigner's composition of the check into its verdict (ADR 0528 clause 5): a failed composition +// fails the check, naming the machine it failed at. A try is not composed (it has no candidate); a composition that +// cannot be had leaves it not run, said. +func (st *state) composed(ctx context.Context, v *delivery.Verdict) { + if st.j.Compositions == nil || st.c.Try != nil { + return + } + comp, err := st.j.Compositions.Await(ctx, st.c.Key) + if err != nil { + st.j.logf("%s: the composition could not be read, so it is said as not run: %v", st.c.Key, err) + return + } + v.Composition = comp.Composition() + v.Took.Composition = comp.Took + if comp.State == delivery.CompositionFail { + machine := comp.Machine + if machine == "" { + machine = "the composition" + } + v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: machine, Words: comp.Words} + } +} + // outcome fills the verdict's levels, times, machines and composition from what ran. func (st *state) outcome(v delivery.Verdict) Outcome { if v.State == delivery.VerdictPass || len(v.Levels) == 0 { diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index a4b5442..0fc6074 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -96,7 +96,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1 t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", - "mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", + "mesh-controller.plans-building", "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) { t.Errorf("reads = %v, want %v", m.Reads, want) } diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 2c22576..1f15b6c 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -22,6 +22,7 @@ "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.plans-building", + "mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", From 016385751d2096701e17c6ffce46975c79e2de6f Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:29:57 +0200 Subject: [PATCH 05/10] =?UTF-8?q?Write=20each=20node's=20want=20by=20the?= =?UTF-8?q?=20six=20rules,=20the=20canary=20the=20first=20online=20node=20?= =?UTF-8?q?by=20name,=20and=20hold=20a=20fingerprint=20whose=20health=20fa?= =?UTF-8?q?iled=20with=20its=20nodes=20set=20back,=20so=20R5's=20wants=20c?= =?UTF-8?q?an=20be=20compared=20with=20what=20the=20controller=20sends=20(?= =?UTF-8?q?issue=201297,=20subtasks=20B4a-B4b;=20ADR=200429=20=C2=A76,=20A?= =?UTF-8?q?DR=200528=20clauses=201=20and=203,=20ADR=200435=20=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-assigner/main.go | 6 +- .../cmd/mesh-delivery-assigner/pass.go | 68 ++++++ .../cmd/mesh-delivery-assigner/world.go | 113 +++++++++- .../internal/assign/wants.go | 205 ++++++++++++++++++ .../internal/assign/wants_test.go | 139 ++++++++++++ 5 files changed, 529 insertions(+), 2 deletions(-) create mode 100644 modules/mesh-delivery-assigner/internal/assign/wants.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/wants_test.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 9bfae45..dc34b09 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -186,7 +186,11 @@ func follow(ctx context.Context, kvs buckets, logger *log.Logger) error { return nil } for _, name := range watched { - if err := watch(name); err != nil { + 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 } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index f5c55fc..8072504 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -5,6 +5,7 @@ import ( "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" @@ -22,6 +23,73 @@ type passer struct { func (p *passer) pass(ctx context.Context) { p.compositions(ctx) + p.wants(ctx) +} + +// 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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index e96c070..d470127 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "slices" "sort" "strings" "sync" @@ -41,6 +42,8 @@ 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 } // parsed is a manifest read by digest: its module and its manifest, or why it cannot be read. @@ -51,7 +54,7 @@ type parsed struct { } func newWorld() *world { - return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}} + return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}} } // put takes a change of a watched bucket in: a value, or nil for a key deleted or purged. @@ -66,6 +69,29 @@ func (w *world) put(bucket, key string, value []byte) { 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. @@ -229,3 +255,88 @@ func (w *world) change(ctx context.Context, manifests getter, c delivery.Candida } 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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants.go b/modules/mesh-delivery-assigner/internal/assign/wants.go new file mode 100644 index 0000000..524f823 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/wants.go @@ -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] +} diff --git a/modules/mesh-delivery-assigner/internal/assign/wants_test.go b/modules/mesh-delivery-assigner/internal/assign/wants_test.go new file mode 100644 index 0000000..c0ff7df --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/wants_test.go @@ -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"]) + } +} From d7ba5aa91e4a6155fd7a5ac8d493c3e9a2d2cb7d Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:30:31 +0200 Subject: [PATCH 06/10] Compare each node's want with the fingerprint of the build the controller sends it, explaining a difference by the rule that holds the want back or saying it unexplained (issue 1297, subtask B5a; ADR 0528 clause 4) --- .../internal/assign/compare.go | 52 ++++++++++++++++ .../internal/assign/compare_test.go | 59 +++++++++++++++++++ 2 files changed, 111 insertions(+) create mode 100644 modules/mesh-delivery-assigner/internal/assign/compare.go create mode 100644 modules/mesh-delivery-assigner/internal/assign/compare_test.go diff --git a/modules/mesh-delivery-assigner/internal/assign/compare.go b/modules/mesh-delivery-assigner/internal/assign/compare.go new file mode 100644 index 0000000..952acd3 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compare.go @@ -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 +} diff --git a/modules/mesh-delivery-assigner/internal/assign/compare_test.go b/modules/mesh-delivery-assigner/internal/assign/compare_test.go new file mode 100644 index 0000000..da4a14f --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/compare_test.go @@ -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) + } + } +} From ed98a7519ee51c86c7f21bbe2712f293d4eb1680 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:31:11 +0200 Subject: [PATCH 07/10] Write every difference between a want and what the controller sends, and delete one that ended, so the view and acceptance test 17 read them on the bus (issue 1297, subtask B5b; ADR 0528 clause 4) --- .../integration_test.go | 62 +++++++++++++++++++ .../cmd/mesh-delivery-assigner/pass.go | 60 +++++++++++++++++- .../cmd/mesh-delivery-assigner/world.go | 30 ++++++++- 3 files changed, 150 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index d45b86a..2682b37 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -212,3 +212,65 @@ func TestAChangeIsComposedOnTheBus(t *testing.T) { 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()) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 8072504..754a812 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -23,7 +23,65 @@ type passer struct { func (p *passer) pass(ctx context.Context) { p.compositions(ctx) - p.wants(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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index d470127..01ad27a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -44,6 +44,8 @@ type world struct { 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. @@ -54,7 +56,7 @@ type parsed struct { } func newWorld() *world { - return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}} + 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. @@ -340,3 +342,29 @@ func (w *world) built(module, fp string) bool { 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 `.build.`) 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 +} From c9ee92a4abd1b15b980f04830a2006ba432a77e0 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:39:49 +0200 Subject: [PATCH 08/10] Pin the assigner and the checker to the SDK commit that holds the assigner's ask value, so both build on the layouts the view reads (issue 1297, subtask B1c) --- modules/mesh-delivery-assigner/go.mod | 2 +- modules/mesh-delivery-checker/go.mod | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 4998342..6c56f76 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index e42ae23..49477b2 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) From 7cbb1c4207456a9adf2f36ce540f7fbbb1c8d9cf Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:44:00 +0200 Subject: [PATCH 09/10] Wait for the assigner's composition only while the assigner is online, say at once that it is not judged when it is not, and take it in when the assigner writes it later, so no check waits its bound for a composition nobody makes (issue 1297, B3d) --- .../cmd/mesh-delivery-checker/compositions.go | 100 +++++++++--- .../compositions_integration_test.go | 26 ++- .../cmd/mesh-delivery-checker/main.go | 9 +- modules/mesh-delivery-checker/go.mod | 2 +- .../internal/check/awaiting.go | 141 +++++++++++++++++ .../internal/check/awaiting_test.go | 149 ++++++++++++++++++ .../internal/check/judge.go | 19 +-- .../internal/check/line.go | 3 + 8 files changed, 412 insertions(+), 37 deletions(-) create mode 100644 modules/mesh-delivery-checker/internal/check/awaiting.go create mode 100644 modules/mesh-delivery-checker/internal/check/awaiting_test.go diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go index 951b0a6..5770315 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions.go @@ -2,44 +2,98 @@ package main import ( "context" - "errors" - "fmt" "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: no poll and no -// timer, the check's own bound ends the wait. -type compositionWatch struct{ kv jetstream.KeyValue } +// 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; a value it cannot read is an -// error, and ctx ending is ctx's error. +// 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) { - w, err := c.kv.Watch(ctx, key) + 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 w.Stop() - for { - select { - case <-ctx.Done(): - return delivery.CompositionVerdict{}, ctx.Err() - case e, ok := <-w.Updates(): - if !ok { - return delivery.CompositionVerdict{}, errors.New("the watch of the composition ended") + 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 } - v, err := delivery.DecodeCompositionVerdict(e.Value()) - if err != nil { - return delivery.CompositionVerdict{}, fmt.Errorf("the composition %s cannot be read: %w", key, err) - } - if v.State != delivery.CompositionWaits { - return v, nil + if comp, err := delivery.DecodeCompositionVerdict(e.Value()); err == nil { + ch.Composed(ctx, comp) } } - } + }() + return nil } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go index 070abae..0325a35 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/compositions_integration_test.go @@ -4,6 +4,7 @@ package main import ( "context" + "errors" "os" "path/filepath" "testing" @@ -14,6 +15,8 @@ import ( "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 @@ -45,10 +48,31 @@ func TestTheCheckerAwaitsTheAssignersComposition(t *testing.T) { 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}.Await(ctx, key) + v, err := compositionWatch{kv: kv, presence: pkv}.Await(ctx, key) if err != nil { t.Error(err) } diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index 676d380..08b3c0b 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -231,7 +231,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro case err != nil: return fmt.Errorf("the bucket %s: %w", delivery.AssignerCompositionsBucket, err) default: - compositions = compositionWatch{kv} + 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} diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 49477b2..53b499f 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4 + 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 ) diff --git a/modules/mesh-delivery-checker/internal/check/awaiting.go b/modules/mesh-delivery-checker/internal/check/awaiting.go new file mode 100644 index 0000000..1fbae2c --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting.go @@ -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.`). 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) +} diff --git a/modules/mesh-delivery-checker/internal/check/awaiting_test.go b/modules/mesh-delivery-checker/internal/check/awaiting_test.go new file mode 100644 index 0000000..c775c49 --- /dev/null +++ b/modules/mesh-delivery-checker/internal/check/awaiting_test.go @@ -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 } diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index 099caca..d7e420c 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -3,6 +3,7 @@ package check import ( "context" "encoding/json" + "errors" "fmt" "path" "slices" @@ -253,25 +254,21 @@ func (st *state) finish(ctx context.Context, v delivery.Verdict) Outcome { // 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. +// 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) - if err != nil { + 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 } - 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} - } + withComposition(v, comp) } // outcome fills the verdict's levels, times, machines and composition from what ran. diff --git a/modules/mesh-delivery-checker/internal/check/line.go b/modules/mesh-delivery-checker/internal/check/line.go index 85e6830..1030f52 100644 --- a/modules/mesh-delivery-checker/internal/check/line.go +++ b/modules/mesh-delivery-checker/internal/check/line.go @@ -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" } From d5868c178f9f4420d65ba925b74eb1706228b6e8 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:44:00 +0200 Subject: [PATCH 10/10] Take the words of a module the change removes from mesh-sdk in the indexer and the assigner, so the assigner's reading of a removal cannot drift from what the indexer writes (issue 1297, B3b) --- .../cmd/mesh-delivery-assigner/world.go | 4 ++-- modules/mesh-delivery-assigner/go.mod | 2 +- modules/mesh-delivery-indexer/go.mod | 2 +- modules/mesh-delivery-indexer/internal/index/previews.go | 5 +++-- 4 files changed, 7 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 01ad27a..fcd91d8 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -21,8 +21,8 @@ import ( ) // noModuleWords is the indexer's reason for a module of a candidate whose directory the merge preview holds no manifest -// for (mesh-delivery-indexer internal/index/previews.go NonePreviewNoModule): the change removes the module. -const noModuleWords = "the merge preview names no manifest for the module's directory" +// 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. diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index 6c56f76..dbf6452 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023132058-f625d98713e4 + 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 ) diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 6390b2e..47ca71f 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -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 ) diff --git a/modules/mesh-delivery-indexer/internal/index/previews.go b/modules/mesh-delivery-indexer/internal/index/previews.go index 5e8b454..d22a762 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews.go +++ b/modules/mesh-delivery-indexer/internal/index/previews.go @@ -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.