From 71bb5a5fb39a3ebd691c4b98c5e3467de0b9d5fa Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 21 Oct 2026 11:21:46 +0200 Subject: [PATCH] Watch the build modules' toolchains and steps and keep the toolchains by node, with no timer (issue 1607, AC6) --- .../cmd/mesh-delivery-builder/bus.go | 151 ++++++++++++++++++ .../cmd/mesh-delivery-builder/toolchains.go | 104 ++++++++++++ .../internal/build/manifests.go | 12 ++ .../internal/build/toolchains.go | 64 ++++++++ .../internal/build/toolchains_test.go | 51 ++++++ 5 files changed, 382 insertions(+) create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go create mode 100644 modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go create mode 100644 modules/mesh-delivery-builder/internal/build/manifests.go create mode 100644 modules/mesh-delivery-builder/internal/build/toolchains.go create mode 100644 modules/mesh-delivery-builder/internal/build/toolchains_test.go diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go new file mode 100644 index 0000000..5798b3a --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/bus.go @@ -0,0 +1,151 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-builder/internal/build" +) + +// bucketOpener is the part of the bus the watches use. +type bucketOpener interface { + KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) +} + +// kvStore is the builds bucket as the builder's store: create and compare-and-set. +type kvStore struct{ kv jetstream.KeyValue } + +func conflict(err error) error { + if errors.Is(err, jetstream.ErrKeyExists) || errors.Is(err, jetstream.ErrKeyRevisionMismatch) { + return build.ErrConflict + } + return err +} + +func (s kvStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + e, err := s.kv.Get(ctx, key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return nil, 0, false, nil + } + if err != nil { + return nil, 0, false, err + } + return e.Value(), e.Revision(), true, nil +} + +func (s kvStore) Create(ctx context.Context, key string, value []byte) (uint64, error) { + rev, err := s.kv.Create(ctx, key, value) + return rev, conflict(err) +} + +func (s kvStore) Update(ctx context.Context, key string, value []byte, revision uint64) (uint64, error) { + rev, err := s.kv.Update(ctx, key, value, revision) + return rev, conflict(err) +} + +func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut } + +// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the +// changes that follow. +func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry)) (jetstream.KeyWatcher, error) { + w, err := kv.WatchAll(ctx) + if err != nil { + return nil, err + } + for { + select { + case <-ctx.Done(): + w.Stop() + return nil, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return nil, errors.New("the watch ended") + } + if e == nil { + return w, nil + } + each(e) + } + } +} + +// follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process manager +// starts the program again. +func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error { + defer w.Stop() + for { + select { + case <-ctx.Done(): + return nil + case e, ok := <-w.Updates(): + if !ok { + return errors.New("a watch ended") + } + if e != nil { + each(e) + } + } + } +} + +// stepsWaiter waits for a build module's step by watching its key in the module's steps bucket (ADR 0469 clause 2): the +// watch hands over the value the key holds now, so a step that ended before the watch began is read at once, and every +// change after. No timer: the caller bounds the wait with its context. +type stepsWaiter struct{ js bucketOpener } + +func (w stepsWaiter) Wait(ctx context.Context, module, key string) (delivery.Step, error) { + kv, err := w.js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + watch, err := kv.Watch(ctx, key) + if err != nil { + return delivery.Step{}, err + } + defer watch.Stop() + for { + select { + case <-ctx.Done(): + return delivery.Step{}, ctx.Err() + case e, ok := <-watch.Updates(): + if !ok { + return delivery.Step{}, errors.New("the watch of the step ended") + } + if e == nil || !opOf(e) { + continue + } + step, err := delivery.DecodeStep(e.Value()) + if err != nil { + return delivery.Step{}, fmt.Errorf("the step %s is not a step: %w", key, err) + } + if step.State != delivery.StepRunning { + return step, nil + } + } + } +} + +// manifestReader reads manifests. of the indexer and parses the manifest it holds. +type manifestReader struct{ kv jetstream.KeyValue } + +func (m manifestReader) Manifest(ctx context.Context, digest string) (catalogue.Manifest, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return catalogue.Manifest{}, err + } + e, err := m.kv.Get(ctx, key) + if err != nil { + return catalogue.Manifest{}, fmt.Errorf("manifests.%s: %w", key, err) + } + kept, err := delivery.DecodeManifest(e.Value()) + if err != nil { + return catalogue.Manifest{}, err + } + return catalogue.ParseManifest([]byte(kept.JSON)) +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go new file mode 100644 index 0000000..a7afc4d --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/toolchains.go @@ -0,0 +1,104 @@ +package main + +import ( + "context" + "errors" + "fmt" + "log" + "strings" + "sync" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "mesh-delivery-builder/internal/build" +) + +// toolchainStreams are the stream each build module carries its toolchain.reported event on, as the bus names it +// (declared by the build modules, heard by this module through consumes). +var toolchainStreams = map[string]string{build.GoBuildModule: "TOOLCHAINS_GO", build.NodeBuildModule: "TOOLCHAINS_NODE"} + +// toolchainWatch keeps the build modules' toolchains in a build.ToolchainSet and tells the builder when one changed. A +// build module's bucket is made when it is assigned, after the builder started, and the builder holds no timer (ADR 0437 +// ยง8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key +// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched. +type toolchainWatch struct { + ctx context.Context + js bucketOpener + set *build.ToolchainSet + changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again + logger *log.Logger + errs chan error + mu sync.Mutex + watched map[string]bool +} + +func newToolchainWatch(ctx context.Context, js bucketOpener, set *build.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch { + return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(build.LanguageOf)), watched: map[string]bool{}} +} + +// start watches the buckets that exist now; a missing one is said and waits for its build module's event. +func (w *toolchainWatch) start() error { + for module := range build.LanguageOf { + if err := w.ensure(module); err != nil { + return err + } + } + return nil +} + +func (w *toolchainWatch) ensure(module string) error { + w.mu.Lock() + defer w.mu.Unlock() + if w.watched[module] { + return nil + } + name := delivery.ToolchainsBucket(module) + kv, err := w.js.KeyValue(w.ctx, name) + switch { + case errors.Is(err, jetstream.ErrBucketNotFound): + w.logger.Printf("the bucket %s does not exist yet: it is watched when %s reports a toolchain", name, module) + return nil + case err != nil: + return fmt.Errorf("the bucket %s: %w", name, err) + } + watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) }) + if err != nil { + return err + } + w.watched[module] = true + go func() { + w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) { + if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) { + w.changed(w.ctx) + } + }) + }() + w.changed(w.ctx) // what the settle read is new to the builder + return nil +} + +// delivered is the part of a JetStream message the watch uses. +type delivered interface { + Subject() string + Ack() error + Term() error +} + +// Reported takes one toolchain.reported event of a build module in: the module is read off the subject, +// mesh.stream..event.toolchain.reported, and its bucket is watched if it is not. A subject of no build module is +// terminated; a failure stays unacknowledged, and the server delivers it again. +func (w *toolchainWatch) Reported(m delivered) { + module, _, _ := strings.Cut(strings.TrimPrefix(m.Subject(), "mesh.stream."), ".event.") + if _, known := build.LanguageOf[module]; !known || !strings.HasPrefix(m.Subject(), "mesh.stream.") { + w.logger.Printf("%s: not a build module's event, dropped", m.Subject()) + _ = m.Term() + return + } + if err := w.ensure(module); err != nil { + w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err) + return + } + _ = m.Ack() +} diff --git a/modules/mesh-delivery-builder/internal/build/manifests.go b/modules/mesh-delivery-builder/internal/build/manifests.go new file mode 100644 index 0000000..dbdbc04 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/manifests.go @@ -0,0 +1,12 @@ +package build + +import ( + "context" + + "git.novox.be/novox/mesh-sdk/go/catalogue" +) + +// Manifests reads a module's manifest by its digest, the key of the indexer's manifests bucket. +type Manifests interface { + Manifest(ctx context.Context, digest string) (catalogue.Manifest, error) +} diff --git a/modules/mesh-delivery-builder/internal/build/toolchains.go b/modules/mesh-delivery-builder/internal/build/toolchains.go new file mode 100644 index 0000000..44af6e6 --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/toolchains.go @@ -0,0 +1,64 @@ +package build + +import ( + "sync" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// LanguageOf is the language each build module's toolchains state holds (the indexer's ToolchainModules, reversed). +var LanguageOf = map[string]string{GoBuildModule: "go", NodeBuildModule: "typescript"} + +// ToolchainSet holds what the build modules' toolchains buckets hold: each node's entry of a language, as the watches +// put them in. Safe for use from the watches and from a build at once. +type ToolchainSet struct { + mu sync.Mutex + by map[string]map[string]delivery.Toolchain // language -> node -> entry +} + +// Put takes a change of a build module's toolchains bucket in: a put of a node's entry (key .) or its +// delete. It answers whether the language's entries changed (a node's version or recipe, or a node appearing or going). +// A key or a value that is not an entry of the module's language is left out. +func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool { + language, ok := LanguageOf[module] + if !ok { + return false + } + node, keyLanguage, err := delivery.ParseToolchainKey(key) + if err != nil || keyLanguage != language { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if deleted { + if _, had := s.by[language][node]; !had { + return false + } + delete(s.by[language], node) + return true + } + tc, err := delivery.DecodeToolchain(value) + if err != nil || tc.Language != language { + return false + } + if s.by == nil { + s.by = map[string]map[string]delivery.Toolchain{} + } + if s.by[language] == nil { + s.by[language] = map[string]delivery.Toolchain{} + } + cur, had := s.by[language][node] + s.by[language][node] = tc + return !had || cur.Version != tc.Version || cur.Recipe != tc.Recipe +} + +// Entries are the nodes' entries of a language, in no particular order. +func (s *ToolchainSet) Entries(language string) []delivery.NodeToolchain { + s.mu.Lock() + defer s.mu.Unlock() + var out []delivery.NodeToolchain + for node, tc := range s.by[language] { + out = append(out, delivery.NodeToolchain{Node: node, Toolchain: tc}) + } + return out +} diff --git a/modules/mesh-delivery-builder/internal/build/toolchains_test.go b/modules/mesh-delivery-builder/internal/build/toolchains_test.go new file mode 100644 index 0000000..83b95ca --- /dev/null +++ b/modules/mesh-delivery-builder/internal/build/toolchains_test.go @@ -0,0 +1,51 @@ +package build + +import ( + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +func tcValue(t *testing.T, language, version string) []byte { + t.Helper() + b, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: "1", At: time.Now()}) + if err != nil { + t.Fatal(err) + } + return b +} + +func TestToolchainSetKeepsEachNodesEntryByLanguageAndSaysWhenOneChanged(t *testing.T) { // 1607/AC6 + var s ToolchainSet + if !s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + t.Error("a first entry did not count as a change") + } + if s.Put(GoBuildModule, "novox.go", tcValue(t, "go", "go1.27.1"), false) { + t.Error("the same entry again counted as a change") + } + s.Put(GoBuildModule, "ace.go", tcValue(t, "go", "go1.27.1"), false) + s.Put(NodeBuildModule, "ace.typescript", tcValue(t, "typescript", "tsc5"), false) + if got := s.Entries("go"); len(got) != 2 { + t.Errorf("go entries = %+v, want two", got) + } + if got := s.Entries("typescript"); len(got) != 1 || got[0].Node != "ace" || got[0].Version != "tsc5" { + t.Errorf("typescript entries = %+v", got) + } + if !s.Put(GoBuildModule, "ace.go", nil, true) || len(s.Entries("go")) != 1 { + t.Error("a delete did not remove the node's entry") + } +} + +func TestToolchainSetLeavesOutWhatIsNotAToolchainOfTheModulesLanguage(t *testing.T) { // 1607/AC6 + var s ToolchainSet + if s.Put(GoBuildModule, "novox.typescript", tcValue(t, "typescript", "x"), false) { + t.Error("a typescript toolchain in the go build module's bucket counted") + } + if s.Put(GoBuildModule, "novox.go", []byte("not json"), false) || s.Put("other-module", "novox.go", tcValue(t, "go", "x"), false) { + t.Error("garbage or an unknown module counted") + } + if len(s.Entries("go")) != 0 { + t.Error("an entry was kept") + } +}