Read the builder's keys, the indexer's values by commit and its manifests, so the controller can take a module from a done key (issue 1851, A1, A2)

internal/builderkeys/builderkeys.go

@@ -0,0 +1,278 @@
+// Package builderkeys is how the controller reads what the builder and the indexer wrote (novox/hq ADR 0501, ADR 0498
+// §8, ADR 0453 §5): a module's build at a fingerprint, in the builder's bucket; the indexer's trunk value of a module,
+// which names that fingerprint; and a manifest by its digest. From R4e the controller asks no build but the builder's
+// own module's: it takes every other module from the builder's done key, and it reads them here.
+//
+// It only reads. The builder writes its keys and the indexer its values; the controller's account reaches every
+// bucket through the JetStream API it is granted (internal/broker: KindController).
+package builderkeys
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go/jetstream"
+)
+
+// Key is a module's build at a fingerprint as the builder's bucket holds it. Found is false when there is no key (or
+// no bucket yet): nothing built or queued that fingerprint. Revision is the key's revision, a new one for every
+// rebuild of the same fingerprint (ADR 0429 §18).
+type Key struct {
+	Module      string
+	Fingerprint string
+	Found       bool
+	Revision    uint64
+	Build       delivery.BuilderBuild
+}
+
+// Done says the key is done.
+func (k Key) Done() bool { return k.Found && k.Build.State == delivery.BuildDone }
+
+// Failed says the key failed; its Error is the builder's own words (`digests differ:`, `not stored:`, …).
+func (k Key) Failed() bool { return k.Found && k.Build.State == delivery.BuildFailed }
+
+// Unreferenced are the artifacts of a done key that name no reference (ADR 0498 §8): a key done before R4e's
+// delivery, or a candidate's image or package not yet built again for the trunk. Such a key is waited for, never
+// delivered from.
+func (k Key) Unreferenced() []string {
+	var out []string
+	for _, a := range k.Build.Artifacts {
+		if strings.TrimSpace(a.Reference) == "" {
+			out = append(out, a.Name)
+		}
+	}
+	return out
+}
+
+// Says is the key's state in words: what a plan waiting on it says.
+func (k Key) Says() string {
+	if !k.Found {
+		return fmt.Sprintf("no build of %s at %s yet", k.Module, short(k.Fingerprint))
+	}
+	switch k.Build.State {
+	case delivery.BuildFailed:
+		return fmt.Sprintf("the build of %s at %s failed: %s", k.Module, short(k.Fingerprint), firstLine(k.Build.Error))
+	case delivery.BuildDone:
+		if missing := k.Unreferenced(); len(missing) > 0 {
+			return fmt.Sprintf("the build of %s at %s is done and waits for the reference of %s", k.Module,
+				short(k.Fingerprint), strings.Join(missing, ", "))
+		}
+		return fmt.Sprintf("the build of %s at %s is done", k.Module, short(k.Fingerprint))
+	}
+	if k.Build.By != "" {
+		return fmt.Sprintf("the build of %s at %s is %s on %s", k.Module, short(k.Fingerprint), k.Build.State, k.Build.By)
+	}
+	return fmt.Sprintf("the build of %s at %s is %s", k.Module, short(k.Fingerprint), k.Build.State)
+}
+
+// ErrNoValue is a module the indexer holds no value of at the commit asked: not written yet, or older than the
+// bucket's history.
+var ErrNoValue = errors.New("the indexer holds no value of the module at that commit")
+
+// Reader reads the builder's and the indexer's buckets.
+type Reader struct {
+	JS jetstream.JetStream
+}
+
+// bucket is the named bucket, or nil when it is not there yet (the part that writes it not delivered).
+func (r Reader) bucket(ctx context.Context, name string) (jetstream.KeyValue, error) {
+	kv, err := r.JS.KeyValue(ctx, name)
+	if errors.Is(err, jetstream.ErrBucketNotFound) {
+		return nil, nil
+	}
+	if err != nil {
+		return nil, fmt.Errorf("the bucket %s: %w", name, err)
+	}
+	return kv, nil
+}
+
+// Build is the builder's key of module at fingerprint fp (`src2:` and 64 hex digits). A missing key, or a missing
+// bucket, is a Key not Found and no error; a value that does not decode is an error naming the key.
+func (r Reader) Build(ctx context.Context, module, fp string) (Key, error) {
+	out := Key{Module: module, Fingerprint: fp}
+	key, err := delivery.BuilderBuildKey(module, fp)
+	if err != nil {
+		return out, err
+	}
+	kv, err := r.bucket(ctx, delivery.BuilderBuildsBucket)
+	if err != nil || kv == nil {
+		return out, err
+	}
+	e, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return out, nil
+	}
+	if err != nil {
+		return out, fmt.Errorf("the builder's key %s: %w", key, err)
+	}
+	b, err := delivery.DecodeBuilderBuild(e.Value())
+	if err != nil {
+		return out, fmt.Errorf("the builder's key %s is not read: %w", key, err)
+	}
+	out.Found, out.Revision, out.Build = true, e.Revision(), b
+	return out, nil
+}
+
+// ModuleAt is the indexer's trunk value of module written for commit (ADR 0501 §2): the newest revision of
+// modules.<module> whose Commit is commit, read from the key's history; with commit empty, the value it holds now.
+// None held is ErrNoValue.
+func (r Reader) ModuleAt(ctx context.Context, module, commit string) (delivery.Module, error) {
+	key, err := delivery.ModulesKey(module)
+	if err != nil {
+		return delivery.Module{}, err
+	}
+	kv, err := r.bucket(ctx, delivery.ModulesBucket)
+	if err != nil {
+		return delivery.Module{}, err
+	}
+	if kv == nil {
+		return delivery.Module{}, fmt.Errorf("%w (%s: the indexer's bucket is not there)", ErrNoValue, module)
+	}
+	var values [][]byte
+	if commit == "" {
+		e, err := kv.Get(ctx, key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return delivery.Module{}, fmt.Errorf("%w (%s)", ErrNoValue, module)
+		}
+		if err != nil {
+			return delivery.Module{}, fmt.Errorf("the indexer's value of %s: %w", module, err)
+		}
+		values = [][]byte{e.Value()}
+	} else {
+		entries, err := kv.History(ctx, key)
+		if errors.Is(err, jetstream.ErrKeyNotFound) {
+			return delivery.Module{}, fmt.Errorf("%w (%s at %s)", ErrNoValue, module, short(commit))
+		}
+		if err != nil {
+			return delivery.Module{}, fmt.Errorf("the indexer's values of %s: %w", module, err)
+		}
+		for _, e := range entries {
+			if e.Operation() == jetstream.KeyValuePut {
+				values = append(values, e.Value())
+			}
+		}
+	}
+	return pickAt(module, commit, values)
+}
+
+// pickAt is, of a key's values oldest first, the newest whose commit is commit (any, with commit empty); values that
+// do not decode are passed over, and one of them is said when nothing else is found.
+func pickAt(module, commit string, values [][]byte) (delivery.Module, error) {
+	var unread error
+	for i := len(values) - 1; i >= 0; i-- {
+		m, err := delivery.DecodeModule(values[i])
+		if err != nil {
+			if unread == nil {
+				unread = err
+			}
+			continue
+		}
+		if commit == "" || sameCommit(m.Commit, commit) {
+			return m, nil
+		}
+	}
+	where := module
+	if commit != "" {
+		where += " at " + short(commit)
+	}
+	if unread != nil {
+		return delivery.Module{}, fmt.Errorf("%w (%s; a value is not read: %v)", ErrNoValue, where, unread)
+	}
+	return delivery.Module{}, fmt.Errorf("%w (%s)", ErrNoValue, where)
+}
+
+// sameCommit is two commit ids naming one commit: equal, or one a prefix of the other of at least 7 hex digits.
+func sameCommit(a, b string) bool {
+	if a == b {
+		return a != ""
+	}
+	if len(a) < 7 || len(b) < 7 {
+		return false
+	}
+	return strings.HasPrefix(a, b) || strings.HasPrefix(b, a)
+}
+
+// Manifest is the indexer's manifest of digest: the module and its module.json byte for byte.
+func (r Reader) Manifest(ctx context.Context, digest string) (delivery.Manifest, error) {
+	key, err := delivery.ManifestsKey(digest)
+	if err != nil {
+		return delivery.Manifest{}, err
+	}
+	kv, err := r.bucket(ctx, delivery.ManifestsBucket)
+	if err != nil {
+		return delivery.Manifest{}, err
+	}
+	if kv == nil {
+		return delivery.Manifest{}, fmt.Errorf("the manifest %s: the indexer's manifests bucket is not there", short(digest))
+	}
+	e, err := kv.Get(ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return delivery.Manifest{}, fmt.Errorf("the indexer holds no manifest %s", short(digest))
+	}
+	if err != nil {
+		return delivery.Manifest{}, fmt.Errorf("the manifest %s: %w", short(digest), err)
+	}
+	m, err := delivery.DecodeManifest(e.Value())
+	if err != nil {
+		return delivery.Manifest{}, fmt.Errorf("the manifest %s is not read: %w", short(digest), err)
+	}
+	return m, nil
+}
+
+// Watch calls wake with the module of every key the builder or the indexer writes from now on, until ctx ends: what
+// wakes a plan waiting on a key, with no timer (ADR 0501; the R4e context's A7). A bucket not there yet is watched once
+// it is: the watch looks again each time wake would have been called for the other bucket, and at every retry.
+func Watch(ctx context.Context, js jetstream.JetStream, wake func(module string)) error {
+	for _, name := range []string{delivery.BuilderBuildsBucket, delivery.ModulesBucket} {
+		if err := watchOne(ctx, js, name, wake); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func watchOne(ctx context.Context, js jetstream.JetStream, name string, wake func(module string)) error {
+	kv, err := js.KeyValue(ctx, name)
+	if err != nil {
+		return fmt.Errorf("the bucket %s: %w", name, err)
+	}
+	w, err := kv.WatchAll(ctx, jetstream.UpdatesOnly())
+	if err != nil {
+		return fmt.Errorf("watching %s: %w", name, err)
+	}
+	go func() {
+		defer w.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case e, ok := <-w.Updates():
+				if !ok {
+					return
+				}
+				if e == nil || e.Operation() != jetstream.KeyValuePut {
+					continue
+				}
+				module, _, _ := strings.Cut(e.Key(), ".")
+				wake(module)
+			}
+		}
+	}()
+	return nil
+}
+
+func short(s string) string {
+	s = strings.TrimPrefix(s, "src2:")
+	if len(s) > 12 {
+		return s[:12]
+	}
+	return s
+}
+
+func firstLine(s string) string {
+	line, _, _ := strings.Cut(strings.TrimSpace(s), "\n")
+	return line
+}

internal/builderkeys/builderkeys_integration_test.go

@@ -0,0 +1,138 @@
+//go:build integration
+
+package builderkeys
+
+import (
+	"context"
+	"errors"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/novox/mesh-controller/internal/broker"
+	"github.com/novox/mesh-controller/internal/testbus"
+)
+
+func aBus(t *testing.T) jetstream.JetStream {
+	t.Helper()
+	conn, err := nats.Connect(testbus.URL(t))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(conn.Close)
+	js, err := jetstream.New(conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return js
+}
+
+func put(t *testing.T, kv jetstream.KeyValue, key string, value []byte) {
+	t.Helper()
+	if _, err := kv.Put(context.Background(), key, value); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// The controller reads the builder's keys, the indexer's values by commit and its manifests (A1, A2), and a key the
+// builder writes wakes it with the key's module (A1's watch, no timer).
+func TestTheControllerReadsTheBuildersKeysAndTheIndexersValues(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
+	defer cancel()
+	js := aBus(t)
+	r := Reader{JS: js}
+
+	// Nothing there yet: no bucket is no key and no error; no value is ErrNoValue.
+	if k, err := r.Build(ctx, "x", fp); err != nil || k.Found {
+		t.Fatalf("no bucket: %+v %v", k, err)
+	}
+	if _, err := r.ModuleAt(ctx, "x", "abcdef1"); !errors.Is(err, ErrNoValue) {
+		t.Fatalf("no bucket: %v", err)
+	}
+
+	builds, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.BuilderBuildsBucket,
+		History: delivery.BuilderBuildsHistory})
+	if err != nil {
+		t.Fatal(err)
+	}
+	modules, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ModulesBucket,
+		History: delivery.ModulesHistory})
+	if err != nil {
+		t.Fatal(err)
+	}
+	manifests, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ManifestsBucket})
+	if err != nil {
+		t.Fatal(err)
+	}
+	woke := make(chan string, 8)
+	if err := Watch(ctx, js, func(m string) { woke <- m }); err != nil {
+		t.Fatal(err)
+	}
+
+	raw := `{"module":"x","version":"1"}`
+	digest := delivery.ManifestDigest([]byte(raw))
+	value, err := delivery.EncodeManifest(delivery.Manifest{Module: "x", JSON: raw})
+	if err != nil {
+		t.Fatal(err)
+	}
+	put(t, manifests, digest, value)
+
+	key, _ := delivery.BuilderBuildKey("x", fp)
+	building, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace"})
+	put(t, builds, key, building)
+	select {
+	case m := <-woke:
+		if m != "x" {
+			t.Fatalf("woken for %q", m)
+		}
+	case <-ctx.Done():
+		t.Fatal("the builder's write woke nothing")
+	}
+	done, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, Manifest: digest,
+		Artifacts: []delivery.BuiltArtifact{{Name: "a", Kind: "bundle", Digest: "sha256:1",
+			Reference: "artifact-store://x/a/blobs/sha256:1"}}})
+	put(t, builds, key, done)
+	k, err := r.Build(ctx, "x", fp)
+	if err != nil || !k.Done() || len(k.Unreferenced()) != 0 || k.Build.Artifacts[0].Reference == "" {
+		t.Fatalf("the done key: %+v %v", k, err)
+	}
+	m, err := r.Manifest(ctx, k.Build.Manifest)
+	if err != nil || m.JSON != raw {
+		t.Fatalf("the manifest: %+v %v", m, err)
+	}
+
+	other := "src2:" + strings.Repeat("e", 64)
+	for _, v := range []struct{ commit, fp2 string }{{"1111111111", other}, {"2222222222", fp}, {"3333333333", other}} {
+		raw, err := delivery.EncodeModule(delivery.Module{FP: "src1:" + strings.Repeat("a", 64), FP2: v.fp2,
+			Manifest: digest, Repository: "novox/r", Commit: v.commit})
+		if err != nil {
+			t.Fatal(err)
+		}
+		put(t, modules, "x", raw)
+	}
+	at, err := r.ModuleAt(ctx, "x", "2222222222")
+	if err != nil || at.FP2 != fp {
+		t.Fatalf("FP2 at the plan's commit, overwritten by a later merge: %+v %v", at, err)
+	}
+	if now, err := r.ModuleAt(ctx, "x", ""); err != nil || now.Commit != "3333333333" {
+		t.Fatalf("the value now: %+v %v", now, err)
+	}
+	if _, err := r.ModuleAt(ctx, "x", "9999999999"); !errors.Is(err, ErrNoValue) {
+		t.Fatalf("a commit the indexer never wrote: %v", err)
+	}
+}
+
+// The controller's account reaches the builder's and the indexer's buckets: it is granted the JetStream API whole.
+func TestTheControllersAccountReachesTheBuckets(t *testing.T) {
+	perms, err := broker.PermissionsFor(broker.Principal{Kind: broker.KindController, PasswordHash: "x"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !slices.Contains(perms.Publish, "$JS.API.>") || !slices.Contains(perms.Subscribe, "$JS.API.>") {
+		t.Fatalf("the controller is not granted the JetStream API: %v / %v", perms.Publish, perms.Subscribe)
+	}
+}

internal/builderkeys/builderkeys_test.go

@@ -0,0 +1,86 @@
+package builderkeys
+
+import (
+	"errors"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+const fp = "src2:" + "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef"
+
+func moduleValue(t *testing.T, commit, fp2 string) []byte {
+	t.Helper()
+	raw, err := delivery.EncodeModule(delivery.Module{FP: "src1:" + strings.Repeat("a", 64), FP2: fp2,
+		Manifest: strings.Repeat("b", 64), Repository: "novox/mesh-catalog", Commit: commit})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return raw
+}
+
+// A plan's module is the indexer's value written for the merge the plan carries, not a later merge's (ADR 0501 §2).
+func TestPickAtIsTheNewestValueOfTheCommit(t *testing.T) {
+	other := "src2:" + strings.Repeat("c", 64)
+	values := [][]byte{
+		moduleValue(t, "1111111111", other),
+		moduleValue(t, "2222222222", other),
+		moduleValue(t, "2222222222", fp), // the same merge, computed again when a toolchain appeared
+		moduleValue(t, "3333333333", other),
+	}
+	m, err := pickAt("x", "2222222222", values)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if m.FP2 != fp {
+		t.Fatalf("FP2 at 2222222222 is %s, want the newest of that commit %s", m.FP2, fp)
+	}
+	// An abbreviated commit names the same one.
+	if m, err := pickAt("x", "2222222", values); err != nil || m.FP2 != fp {
+		t.Fatalf("an abbreviated commit: %v %v", m.FP2, err)
+	}
+	// No commit is the newest value.
+	if m, err := pickAt("x", "", values); err != nil || m.Commit != "3333333333" {
+		t.Fatalf("the newest: %v %v", m.Commit, err)
+	}
+}
+
+func TestPickAtSaysNoValueOfACommitNotWritten(t *testing.T) {
+	_, err := pickAt("x", "4444444444", [][]byte{moduleValue(t, "1111111111", fp), []byte("{not json")})
+	if !errors.Is(err, ErrNoValue) {
+		t.Fatalf("want ErrNoValue, got %v", err)
+	}
+	if !strings.Contains(err.Error(), "x at 4444444444") || !strings.Contains(err.Error(), "not read") {
+		t.Fatalf("the words name neither the module and commit nor the unread value: %v", err)
+	}
+}
+
+// A done key with references is delivered from; a done key with an artifact without one is waited for, said
+// (ADR 0498 §8); a failed key says the builder's own words.
+func TestAKeySaysWhatItIs(t *testing.T) {
+	missing := Key{Module: "x", Fingerprint: fp}
+	if missing.Done() || missing.Failed() || !strings.Contains(missing.Says(), "no build of x") {
+		t.Fatalf("a missing key: %q", missing.Says())
+	}
+	done := Key{Module: "x", Fingerprint: fp, Found: true, Build: delivery.BuilderBuild{State: delivery.BuildDone,
+		Artifacts: []delivery.BuiltArtifact{
+			{Name: "a", Digest: "sha256:1", Reference: "artifact-store://x/a/blobs/sha256:1"},
+			{Name: "img", Kind: "image", Digest: "sha256:2"},
+		}}}
+	if !done.Done() || len(done.Unreferenced()) != 1 || done.Unreferenced()[0] != "img" {
+		t.Fatalf("unreferenced: %v", done.Unreferenced())
+	}
+	if !strings.Contains(done.Says(), "waits for the reference of img") {
+		t.Fatalf("a done key without a reference: %q", done.Says())
+	}
+	failed := Key{Module: "x", Fingerprint: fp, Found: true, Build: delivery.BuilderBuild{State: delivery.BuildFailed,
+		Error: "digests differ: ace sha256:1, g14 sha256:2\nmore"}}
+	if !failed.Failed() || !strings.HasSuffix(failed.Says(), "failed: digests differ: ace sha256:1, g14 sha256:2") {
+		t.Fatalf("a failed key: %q", failed.Says())
+	}
+	building := Key{Module: "x", Fingerprint: fp, Found: true, Build: delivery.BuilderBuild{State: delivery.BuildBuilding, By: "ace"}}
+	if building.Done() || !strings.Contains(building.Says(), "building on ace") {
+		t.Fatalf("a building key: %q", building.Says())
+	}
+}