From 1ed7e3415522f50e2738c90f8b0d2f7f23b9ccf1 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 22 Oct 2026 17:28:13 +0200 Subject: [PATCH] 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 | 278 ++++++++++++++++++ .../builderkeys_integration_test.go | 138 +++++++++ internal/builderkeys/builderkeys_test.go | 86 ++++++ 3 files changed, 502 insertions(+) create mode 100644 internal/builderkeys/builderkeys.go create mode 100644 internal/builderkeys/builderkeys_integration_test.go create mode 100644 internal/builderkeys/builderkeys_test.go diff --git a/internal/builderkeys/builderkeys.go b/internal/builderkeys/builderkeys.go new file mode 100644 index 00000000..b931defd --- /dev/null +++ b/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. 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 +} diff --git a/internal/builderkeys/builderkeys_integration_test.go b/internal/builderkeys/builderkeys_integration_test.go new file mode 100644 index 00000000..80352ad1 --- /dev/null +++ b/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) + } +} diff --git a/internal/builderkeys/builderkeys_test.go b/internal/builderkeys/builderkeys_test.go new file mode 100644 index 00000000..47a82a4b --- /dev/null +++ b/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()) + } +}