Watch the build modules' toolchains and steps and keep the toolchains by node, with no timer (issue 1607, AC6)

2026-10-21 11:21:46 +02:00
parent 3298ad58d3
commit 71bb5a5fb3
5 changed files with 382 additions and 0 deletions
@@ -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.<digest> 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))
+}
@@ -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.<module>.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()
+}
@@ -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)
+}
@@ -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 <node>.<language>) 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
+}
@@ -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")
+	}
+}