// 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 }