The composer's Build is made from the catalog half's entry it holds and the parts of the want that names it (buildOf); the inputs of the indexer's modules and the builder's builds, takeBuildDone, takeModule and the builder read are gone, and mesh-host names no BuilderBuildsBucket. The shadow takes the entries of the builds it composes from (askBuildsEntries), builds a module a candidate adds from its entry and the candidate's `added` parts (A-316), and holds, with the reason, a module whose entry has no part given, whose fingerprint the verb module does not give, or for which it errors (A-312). Its composer is given the entry call as the node's own is. Rows: a built module added with and without parts, the verb module answering no fingerprint and erroring. Builds: design 47 G §421, §628, A-308, A-312, A-316
679 lines
24 KiB
Go
679 lines
24 KiB
Go
package selfcompose
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"fmt"
|
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
|
"maps"
|
|
"reflect"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
|
|
"github.com/novox/mesh-host/internal/agent"
|
|
)
|
|
|
|
// **The catalog half, read by fingerprint** (design 47 G §421, §683, §698; the lead's answers A-47 and A-50). The node-engine
|
|
// follows no branch: for each build its wants and the runnings it heard name, it holds the catalog graph's entry of that
|
|
// fingerprint, asked of the seat mesh-catalog's verb `entry` once (meshcatalogclient.Entries) and kept in its composition
|
|
// state `composed-<node>` beside the compositions (`entry.<fingerprint>`, the entry as `entry` answers it), with each want it
|
|
// composes from (`want.<module>`), written whole before the composition that uses them and read at its start before it asks
|
|
// anything. A module's manifest is the entry's, and each artifact's reference is made from the digest its want's parts name
|
|
// (referenceOf), so it reads neither the indexer's states nor the builder's.
|
|
|
|
// The keys of the kept entries and wants in the composition state.
|
|
const (
|
|
keptEntryPrefix = "entry."
|
|
keptWantPrefix = "want."
|
|
// keptWithin bounds one write of the composition state from an answer of the catalog half.
|
|
keptWithin = 15 * time.Second
|
|
)
|
|
|
|
// keptWant is a want as the composition state keeps it (§683): the want beside when this node first kept it.
|
|
type keptWant struct {
|
|
V int `json:"v"`
|
|
Since time.Time `json:"since"`
|
|
Want meshwants.WantsByWantNodeModuleValue `json:"want"`
|
|
}
|
|
|
|
// keptEntry is an entry held from the composition state or an answer, with when it was kept.
|
|
type keptEntry struct {
|
|
entry meshcatalog.CatalogEntry
|
|
since time.Time
|
|
}
|
|
|
|
// EntryRow is one entry the node-engine keeps, as the seat's verb first-views lists it (§698).
|
|
type EntryRow struct {
|
|
Fingerprint string `json:"fingerprint"`
|
|
State string `json:"state"`
|
|
Since time.Time `json:"since"`
|
|
}
|
|
|
|
// WantRow is one want the node-engine keeps, as the seat's verb first-views lists it (§698).
|
|
type WantRow struct {
|
|
Module string `json:"module"`
|
|
Fingerprint string `json:"fingerprint"`
|
|
KeptSince time.Time `json:"kept-since"`
|
|
}
|
|
|
|
// byEntries says the composer reads the catalog half by fingerprint (the node's own composer), as the shadow, which keeps
|
|
// its reads of the indexer until it goes with the checker (§628), does not.
|
|
func (c *Composer) byEntries() bool { return c.entries != nil }
|
|
|
|
// newEntries makes the composer's cache of entries over env's call of `entry`.
|
|
func (c *Composer) newEntries() {
|
|
if c.env.Entry == nil {
|
|
return
|
|
}
|
|
c.entries = meshcatalogclient.NewEntries(c.env.Entry)
|
|
c.entries.Now = c.now
|
|
c.entries.Changed = c.entryChanged
|
|
c.keptEnt = map[string]keptEntry{}
|
|
c.keptWants = map[string]bool{}
|
|
c.keptWantVal = map[string]keptWant{}
|
|
}
|
|
|
|
// entryKey is an entry's key in the composition state, wantKey a want's.
|
|
//
|
|
// The fingerprint is keyed by its digits, as the compositions' keys are (engcomp.Key): a key holds no colon.
|
|
func entryKey(fp string) string {
|
|
return keptEntryPrefix + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2)
|
|
}
|
|
|
|
func wantStateKey(module string) string { return keptWantPrefix + module }
|
|
|
|
// loadKept reads the entries and wants the composition state keeps, once at the start, before anything is asked of the catalog
|
|
// half (§683). A want read stands only until the live read of the wants answers (endKeptWants). A state that cannot be read
|
|
// is said, and what it would have kept is asked of `entry` and read from the wants as they come.
|
|
func (c *Composer) loadKept(ctx context.Context) {
|
|
if !c.byEntries() || c.env.ComposedState == nil {
|
|
return
|
|
}
|
|
kept, err := c.env.ComposedState.Entries(ctx)
|
|
if err != nil {
|
|
c.sayOnceEach("the entries and wants this node kept cannot be read, so each entry its wants name is asked of the catalog half: " + err.Error())
|
|
return
|
|
}
|
|
now := c.now()
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
for key, e := range kept {
|
|
switch {
|
|
case strings.HasPrefix(key, keptEntryPrefix):
|
|
var a meshcatalog.EntryAnswer
|
|
if json.Unmarshal(e.Value, &a) != nil || a.Entry.Module == "" {
|
|
c.env.Say("the kept entry " + key + " cannot be read, and is asked of the catalog half again")
|
|
continue
|
|
}
|
|
c.keptEnt[meshcatalogclient.SourcePrefix2+strings.TrimPrefix(key, keptEntryPrefix)] = keptEntry{entry: a.Entry, since: now}
|
|
case strings.HasPrefix(key, keptWantPrefix):
|
|
var w keptWant
|
|
module := strings.TrimPrefix(key, keptWantPrefix)
|
|
if json.Unmarshal(e.Value, &w) != nil || w.Want.Fp == "" {
|
|
c.env.Say("the kept want " + key + " cannot be read, so it stands until the wants are read")
|
|
continue
|
|
}
|
|
if _, live := c.wants[module]; !live {
|
|
c.wants[module] = w.Want
|
|
c.keptWants[module] = true
|
|
}
|
|
c.keptWantVal[module] = w
|
|
}
|
|
}
|
|
}
|
|
|
|
// endKeptWants ends the standing of the kept wants once the live read of the wants answered: from then the wants the seat
|
|
// says are the ones the node composes from, never a want older than its stream's (§683). A kept want the read did not
|
|
// announce is gone.
|
|
func (c *Composer) endKeptWants() {
|
|
if !c.byEntries() {
|
|
return
|
|
}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
for module := range c.keptWants {
|
|
delete(c.wants, module)
|
|
delete(c.keptWants, module)
|
|
}
|
|
}
|
|
|
|
// entryChanged is told of each entry answered or found missing: an answer is kept in the composition state before the pass it
|
|
// asks (so the composition that uses it finds it kept), and a pass is asked.
|
|
func (c *Composer) entryChanged(row meshcatalogclient.EntryRow) {
|
|
if row.State == meshcatalogclient.EntryAnswered && row.Entry != nil {
|
|
c.mu.Lock()
|
|
c.keptEnt[row.Fingerprint] = keptEntry{entry: *row.Entry, since: row.Since}
|
|
c.mu.Unlock()
|
|
c.keep(entryKey(row.Fingerprint), meshcatalog.EntryAnswer{Entry: *row.Entry})
|
|
}
|
|
c.tellNotFound(row.Fingerprint, row.State == meshcatalogclient.EntryNotFound)
|
|
c.askRehand()
|
|
}
|
|
|
|
// tellNotFound tells the finding entry-not-found once when fp's entry is not found, and once when it no longer is (design 47 G §614, A-117).
|
|
func (c *Composer) tellNotFound(fp string, notFound bool) {
|
|
c.mu.Lock()
|
|
was := c.notFound[fp]
|
|
if was == notFound {
|
|
c.mu.Unlock()
|
|
return
|
|
}
|
|
if c.notFound == nil {
|
|
c.notFound = map[string]bool{}
|
|
}
|
|
if notFound {
|
|
c.notFound[fp] = true
|
|
} else {
|
|
delete(c.notFound, fp)
|
|
}
|
|
c.mu.Unlock()
|
|
if c.env.EntryNotFound != nil {
|
|
c.env.EntryNotFound(fp, notFound)
|
|
}
|
|
}
|
|
|
|
// keep writes one value into the composition state; a state that cannot be written is said, and the value is kept again by
|
|
// the next pass.
|
|
func (c *Composer) keep(key string, v any) {
|
|
if c.env.ComposedState == nil {
|
|
return
|
|
}
|
|
raw, err := json.Marshal(v)
|
|
if err != nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), keptWithin)
|
|
defer cancel()
|
|
c.keptWrite.Lock()
|
|
defer c.keptWrite.Unlock()
|
|
if _, err := c.env.ComposedState.Put(ctx, key, raw); err != nil {
|
|
c.sayOnceEach("the node's composition state cannot keep " + key + ": " + err.Error())
|
|
}
|
|
}
|
|
|
|
// forget deletes one key from the composition state.
|
|
func (c *Composer) forget(key string) {
|
|
if c.env.ComposedState == nil {
|
|
return
|
|
}
|
|
ctx, cancel := context.WithTimeout(context.Background(), keptWithin)
|
|
defer cancel()
|
|
c.keptWrite.Lock()
|
|
defer c.keptWrite.Unlock()
|
|
if err := c.env.ComposedState.Delete(ctx, key); err != nil {
|
|
c.sayOnceEach("the node's composition state cannot drop " + key + ": " + err.Error())
|
|
}
|
|
}
|
|
|
|
// entryOf is the entry of fp as the composer holds it: kept, or asked of the catalog half once (Entries.Need).
|
|
func (c *Composer) entryOf(fp string) (meshcatalog.CatalogEntry, meshcatalogclient.EntryState) {
|
|
c.mu.Lock()
|
|
k, ok := c.keptEnt[fp]
|
|
c.mu.Unlock()
|
|
if ok {
|
|
return k.entry, meshcatalogclient.EntryAnswered
|
|
}
|
|
return c.entries.Need(context.Background(), fp)
|
|
}
|
|
|
|
// wantedBuilds is each fingerprint a want of this node or a running heard names, of the new form, with whether a want names it.
|
|
func (c *Composer) wantedBuilds() map[string]bool {
|
|
out := map[string]bool{}
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
for module, w := range c.wants {
|
|
if module != agent.NodeModule && strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) {
|
|
out[w.Fp] = true
|
|
}
|
|
}
|
|
for _, mods := range c.heard {
|
|
for _, run := range mods {
|
|
if !run.stopped && strings.HasPrefix(run.fingerprint, meshcatalogclient.SourcePrefix2) {
|
|
if _, named := out[run.fingerprint]; !named {
|
|
out[run.fingerprint] = false
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// syncEntries asks the entry of each build a want or a running names, drops the one nothing names any more, keeps each want it
|
|
// composes from, and lays the picture frombus reads (c.modules, c.manifests) from the entries it holds. It runs at the start of
|
|
// each pass.
|
|
func (c *Composer) syncEntries(ctx context.Context) {
|
|
if !c.byEntries() {
|
|
return
|
|
}
|
|
need := c.wantedBuilds()
|
|
for fp := range need {
|
|
c.entryOf(fp)
|
|
}
|
|
for _, row := range c.entries.Rows() {
|
|
c.mu.Lock()
|
|
added := c.addedFps[row.Fingerprint]
|
|
c.mu.Unlock()
|
|
if _, named := need[row.Fingerprint]; !named && !added {
|
|
c.entries.Release(row.Fingerprint)
|
|
c.tellNotFound(row.Fingerprint, false)
|
|
}
|
|
}
|
|
c.mu.Lock()
|
|
var drop []string
|
|
for fp := range c.keptEnt {
|
|
if _, named := need[fp]; !named && !c.addedFps[fp] {
|
|
delete(c.keptEnt, fp)
|
|
drop = append(drop, entryKey(fp))
|
|
}
|
|
}
|
|
type pending struct {
|
|
module string
|
|
kw keptWant
|
|
}
|
|
var put []pending
|
|
var gone []string
|
|
now := c.now()
|
|
for module, w := range c.wants {
|
|
if module == agent.NodeModule || c.keptWants[module] {
|
|
continue
|
|
}
|
|
had, ok := c.keptWantVal[module]
|
|
if ok && reflect.DeepEqual(had.Want, w) {
|
|
continue
|
|
}
|
|
since := now
|
|
if ok && had.Want.Fp == w.Fp {
|
|
since = had.Since
|
|
}
|
|
kw := keptWant{V: 1, Since: since, Want: w}
|
|
c.keptWantVal[module] = kw
|
|
put = append(put, pending{module, kw})
|
|
}
|
|
for module := range c.keptWantVal {
|
|
if _, held := c.wants[module]; !held {
|
|
delete(c.keptWantVal, module)
|
|
gone = append(gone, wantStateKey(module))
|
|
}
|
|
}
|
|
c.mu.Unlock()
|
|
// Written whole before the composition that uses them (§683).
|
|
for _, p := range put {
|
|
c.keep(wantStateKey(p.module), p.kw)
|
|
}
|
|
for _, key := range append(drop, gone...) {
|
|
c.forget(key)
|
|
}
|
|
c.layModules()
|
|
}
|
|
|
|
// layModules lays c.modules and c.manifests from the entries held: what each node was heard to run first, then each module this
|
|
// node's want names, at the want's fingerprint (a want wins).
|
|
func (c *Composer) layModules() {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
modules := map[string]meshcatalog.ModulesByModuleValue{}
|
|
lay := func(module, fp string) {
|
|
k, ok := c.keptEnt[fp]
|
|
if !ok || k.entry.Module == "" {
|
|
return
|
|
}
|
|
e := k.entry
|
|
digest, raw, err := entryManifest(e)
|
|
if err != nil {
|
|
return
|
|
}
|
|
m := meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: e.Repository, Commit: e.Commit, Path: e.Path,
|
|
Needs: e.Needs, Packages: e.Packages, Verdict: e.Verdict}
|
|
if e.Inputs != nil {
|
|
m.Problems, m.Warnings, m.Unknown = e.Inputs.Problems, e.Inputs.Warnings, e.Inputs.Unknown
|
|
}
|
|
modules[module] = m
|
|
c.manifests[digest] = raw
|
|
}
|
|
nodes := make([]string, 0, len(c.heard))
|
|
for node := range c.heard {
|
|
nodes = append(nodes, node)
|
|
}
|
|
sort.Strings(nodes)
|
|
for _, node := range nodes {
|
|
for module, run := range c.heard[node] {
|
|
if !run.stopped && run.fingerprint != "" {
|
|
lay(module, run.fingerprint)
|
|
}
|
|
}
|
|
}
|
|
for module, w := range c.wants {
|
|
if module != agent.NodeModule && w.Fp != "" {
|
|
lay(module, w.Fp)
|
|
}
|
|
}
|
|
// A module an assignment places whose entry is not held (not wanted here, not heard running, or its entry asked and not
|
|
// answered) stands in the catalogue as its name alone, so the placement still finds it: a module held for its entry offers
|
|
// nothing to the rest until its entry is held (§421).
|
|
for _, p := range c.assignments {
|
|
for _, pm := range p.Modules {
|
|
if _, held := modules[pm.Module]; held || pm.Module == "" {
|
|
continue
|
|
}
|
|
raw, _ := json.Marshal(map[string]any{"module": pm.Module, "version": "0"})
|
|
sum := sha256.Sum256(raw)
|
|
digest := "sha256:" + hex.EncodeToString(sum[:])
|
|
// At the fingerprint its want names, so it is no build behind the catalog's and its hold (entryHolds) is what says it.
|
|
modules[pm.Module] = meshcatalog.ModulesByModuleValue{Manifest: digest, Fp2: c.wants[pm.Module].Fp}
|
|
c.manifests[digest] = raw
|
|
}
|
|
}
|
|
c.modules = modules
|
|
}
|
|
|
|
// entryManifest is an entry's manifest as a reader takes it, with its digest: the one the indexer set at take-in, or, where it set
|
|
// none, the digest of the manifest's bytes.
|
|
func entryManifest(e meshcatalog.CatalogEntry) (digest string, raw []byte, err error) {
|
|
if raw, err = json.Marshal(e.Manifest); err != nil {
|
|
return "", nil, err
|
|
}
|
|
digest = e.ManifestDigest
|
|
if digest == "" {
|
|
sum := sha256.Sum256(raw)
|
|
digest = "sha256:" + hex.EncodeToString(sum[:])
|
|
}
|
|
return digest, raw, nil
|
|
}
|
|
|
|
// entryHolds is, of each module this node's want names, why its entry is not held: asked and not answered, or answered not
|
|
// found. It is the phase held with a detail opening entry-asked: or entry-missing: and naming the fingerprint (§421, §614).
|
|
func (c *Composer) entryHolds() map[string]string {
|
|
if !c.byEntries() {
|
|
return nil
|
|
}
|
|
c.mu.Lock()
|
|
wants := map[string]meshwants.WantsByWantNodeModuleValue{}
|
|
for module, w := range c.wants {
|
|
if module != agent.NodeModule {
|
|
wants[module] = w
|
|
}
|
|
}
|
|
c.mu.Unlock()
|
|
out := map[string]string{}
|
|
for module, w := range wants {
|
|
if !strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) {
|
|
continue
|
|
}
|
|
switch e, state := c.entryOf(w.Fp); state {
|
|
case meshcatalogclient.EntryAsked:
|
|
out[module] = fmt.Sprintf("%s the entry of the build %s is asked of the catalog half and not yet answered, so the node composes nothing new of %s",
|
|
agent.EntryAsked, w.Fp, module)
|
|
case meshcatalogclient.EntryNotFound:
|
|
out[module] = fmt.Sprintf("%s the catalog half holds no entry of the build %s, so the node composes nothing new of %s",
|
|
agent.EntryMissing, w.Fp, module)
|
|
default:
|
|
if missing := partsMissing(e, w); missing != "" {
|
|
out[module] = fmt.Sprintf("the want of %s names no digest for %s, so the node composes nothing new of it", module, missing)
|
|
}
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// partsMissing names the artifacts of e whose digest w's parts do not give; "" when each has one.
|
|
func partsMissing(e meshcatalog.CatalogEntry, w meshwants.WantsByWantNodeModuleValue) string {
|
|
var out []string
|
|
for _, a := range e.Artifacts {
|
|
if w.Parts[a.Ref] == "" {
|
|
out = append(out, a.Ref)
|
|
}
|
|
}
|
|
sort.Strings(out)
|
|
return strings.Join(out, ", ")
|
|
}
|
|
|
|
// referenceOf is an artifact's reference made from the digest its want's part names, in the form the artifact store is
|
|
// addressed by (the lead's answer A-50): artifact-store://<module>/<artifact>/blobs/<digest>.
|
|
func referenceOf(module, artifact, digest string) string {
|
|
return "artifact-store://" + module + "/" + artifact + "/blobs/" + digest
|
|
}
|
|
|
|
// buildOf is the build a want names, made from the entry the composer holds and the parts of the want (the builder's done key is
|
|
// not read, §421, §628): done, with each artifact of the entry at the digest its part names and the reference made from it. Nil
|
|
// where the entry is not held, the want names another fingerprint, or a part is missing. The shadow's wants for the modules a
|
|
// candidate adds carry the parts the candidate gives (A-316).
|
|
func (c *Composer) buildOf(module, fp string) *builds.BuilderBuild {
|
|
c.mu.Lock()
|
|
w, wanted := c.wants[module]
|
|
k, ok := c.keptEnt[fp]
|
|
c.mu.Unlock()
|
|
if !wanted || w.Fp != fp || !ok || partsMissing(k.entry, w) != "" {
|
|
return nil
|
|
}
|
|
digest, raw, err := entryManifest(k.entry)
|
|
if err != nil {
|
|
return nil
|
|
}
|
|
c.mu.Lock()
|
|
c.manifests[digest] = raw
|
|
c.mu.Unlock()
|
|
b := builds.BuilderBuild{State: builds.BuildDone, Manifest: digest}
|
|
for _, a := range k.entry.Artifacts {
|
|
_, name, _ := strings.Cut(a.Ref, ":")
|
|
digest := w.Parts[a.Ref]
|
|
b.Artifacts = append(b.Artifacts, builds.BuiltArtifact{Name: name, Kind: a.Kind, Digest: digest,
|
|
Reference: referenceOf(module, name, digest)})
|
|
}
|
|
return &b
|
|
}
|
|
|
|
// FirstViews is the node-engine's first views as the seat's verb first-views lists them (§698): the views it follows (none: it
|
|
// follows no branch), the entries it keeps and the wants it composes from.
|
|
func (c *Composer) FirstViews() (views []map[string]any, entries []EntryRow, wants []WantRow) {
|
|
views, entries, wants = []map[string]any{}, []EntryRow{}, []WantRow{}
|
|
if !c.byEntries() {
|
|
return
|
|
}
|
|
seen := map[string]bool{}
|
|
for _, r := range c.entries.Rows() {
|
|
seen[r.Fingerprint] = true
|
|
entries = append(entries, EntryRow{Fingerprint: r.Fingerprint, State: string(r.State), Since: r.Since})
|
|
}
|
|
c.mu.Lock()
|
|
for fp, k := range c.keptEnt {
|
|
if !seen[fp] {
|
|
entries = append(entries, EntryRow{Fingerprint: fp, State: string(meshcatalogclient.EntryAnswered), Since: k.since})
|
|
}
|
|
}
|
|
for module, w := range c.wants {
|
|
if module == agent.NodeModule || w.Fp == "" {
|
|
continue
|
|
}
|
|
wants = append(wants, WantRow{Module: module, Fingerprint: w.Fp, KeptSince: c.keptWantVal[module].Since})
|
|
}
|
|
c.mu.Unlock()
|
|
sort.Slice(entries, func(i, j int) bool { return entries[i].Fingerprint < entries[j].Fingerprint })
|
|
sort.Slice(wants, func(i, j int) bool { return wants[i].Module < wants[j].Module })
|
|
return
|
|
}
|
|
|
|
// EntryAgain asks fp's entry of the catalog half again now (the seat's verb entry-again, §698): one the composer keeps, whatever
|
|
// its state. The entry kept of it is dropped, so a pass holds the module until the answer.
|
|
func (c *Composer) EntryAgain(ctx context.Context, fp string) error {
|
|
if !c.byEntries() {
|
|
return fmt.Errorf("this engine reads no entry by fingerprint")
|
|
}
|
|
c.mu.Lock()
|
|
_, kept := c.keptEnt[fp]
|
|
delete(c.keptEnt, fp)
|
|
c.mu.Unlock()
|
|
held := kept
|
|
for _, r := range c.entries.Rows() {
|
|
held = held || r.Fingerprint == fp
|
|
}
|
|
if !held {
|
|
return fmt.Errorf("no entry of %s is kept", fp)
|
|
}
|
|
if !kept {
|
|
return c.entries.Again(ctx, fp)
|
|
}
|
|
c.entries.Need(ctx, fp) // makes the row and asks once
|
|
c.askRehand()
|
|
return nil
|
|
}
|
|
|
|
// EntryCall is env.Entry over the engine's link session: the seat mesh-catalog's verb `entry`, read in the engine's own name
|
|
// (the manifest's invokes names it). A refusal in the verb's own words that it keeps no entry of the build is the holder's
|
|
// answer that it holds none, and is told so; any other failure is a call that did not reach the holder, and is told so too.
|
|
func EntryCall(ask func(ctx context.Context, seat, verb string, args any) (json.RawMessage, error)) meshcatalogclient.Call {
|
|
return func(key string, body any) (json.RawMessage, error) {
|
|
seatName, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
raw, err := ask(ctx, seatName, verb, body)
|
|
if err != nil && strings.Contains(err.Error(), noEntryWords) {
|
|
out, _ := json.Marshal(map[string]string{"error": err.Error()})
|
|
return out, nil
|
|
}
|
|
return raw, err
|
|
}
|
|
}
|
|
|
|
// noEntryWords are the words the seat's holder refuses an unknown fingerprint in.
|
|
const noEntryWords = "no entry of fingerprint"
|
|
|
|
// FormChanges are the kept data whose form differs between the manifests of the build a module runs and the one a want moves it
|
|
// to, each read from the entry the composer holds of its fingerprint (design 47 G §296, §297). An entry not held is an error: the
|
|
// node agent then says it is not known and the move goes on.
|
|
func (c *Composer) FormChanges(_ context.Context, module, runningFp, appliedFp string) ([]catalogue.FormChange, error) {
|
|
if !c.byEntries() {
|
|
return nil, fmt.Errorf("this engine reads no entry by fingerprint")
|
|
}
|
|
read := func(fp string) (catalogue.Manifest, error) {
|
|
e, state := c.entryOf(fp)
|
|
if state != meshcatalogclient.EntryAnswered {
|
|
return catalogue.Manifest{}, fmt.Errorf("the entry of %s is %s", fp, state)
|
|
}
|
|
_, raw, err := entryManifest(e)
|
|
if err != nil {
|
|
return catalogue.Manifest{}, err
|
|
}
|
|
return catalogue.Decode(raw)
|
|
}
|
|
running, err := read(runningFp)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s's running manifest: %w", module, err)
|
|
}
|
|
applied, err := read(appliedFp)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s's wanted manifest: %w", module, err)
|
|
}
|
|
return catalogue.FormChanges(running, applied), nil
|
|
}
|
|
|
|
// syncShadowEntries asks the entry of each build a want of this node names, drops the one nothing names, lays c.modules and
|
|
// c.manifests from the entries held, and says whether an entry asked is not yet answered: the shadow's picture of the catalog,
|
|
// where it once watched the indexer's modules (design 47 G §421, A-308).
|
|
func (c *Composer) syncShadowEntries(ctx context.Context) (pending bool) {
|
|
if !c.byEntries() {
|
|
return false
|
|
}
|
|
c.syncEntries(ctx)
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
for module, w := range c.wants {
|
|
if module == agent.NodeModule || !strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) {
|
|
continue
|
|
}
|
|
if _, held := c.keptEnt[w.Fp]; !held {
|
|
for _, r := range c.entries.Rows() {
|
|
if r.Fingerprint == w.Fp && r.State == meshcatalogclient.EntryAsked {
|
|
return true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// moduleVerb is the seat mesh-catalog's verb `module`: one module's trunk fingerprint and manifest by name.
|
|
const moduleVerb = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbModule
|
|
|
|
// addedModule is the trunk's build of a module an assignment candidate adds to this node when no want here names it yet: the
|
|
// fingerprint the seat's verb `module` answers, then its entry by the verb `entry` as for a wanted module (design 47 G §421, A-312).
|
|
// A call, not a watch: nothing of the catalog is kept. It answers the entry when held; otherwise why not: a module with no
|
|
// fingerprint says the verb's reason, an entry asked says so (its answer wakes the shadow, which composes again).
|
|
func (c *Composer) addedModule(ctx context.Context, module string) (fp string, entry meshcatalog.CatalogEntry, held, pending bool, why string) {
|
|
if c.env.Entry == nil {
|
|
return "", entry, false, false, "this node reads no module of the catalog half"
|
|
}
|
|
raw, err := c.env.Entry(moduleVerb, meshcatalog.ModuleInput{Module: module})
|
|
if err != nil {
|
|
return "", entry, false, false, fmt.Sprintf("the catalog half's answer for %s: %v", module, err)
|
|
}
|
|
answer, said, ok := buscalls.Unwrap(raw)
|
|
if !ok {
|
|
return "", entry, false, false, fmt.Sprintf("the catalog half holds no module %s: %s", module, said)
|
|
}
|
|
var a meshcatalog.ModuleAnswer
|
|
if err := json.Unmarshal(answer, &a); err != nil {
|
|
return "", entry, false, false, fmt.Sprintf("the catalog half's answer for %s cannot be read: %v", module, err)
|
|
}
|
|
if a.Module.Fp == "" {
|
|
return "", entry, false, false, fmt.Sprintf("the catalog half gives %s no fingerprint: %s", module, a.Module.None)
|
|
}
|
|
fp = a.Module.Fp
|
|
c.mu.Lock()
|
|
if c.addedFps == nil {
|
|
c.addedFps = map[string]bool{}
|
|
}
|
|
c.addedFps[fp] = true
|
|
c.mu.Unlock()
|
|
e, state := c.entryOf(fp)
|
|
switch state {
|
|
case meshcatalogclient.EntryAnswered:
|
|
return fp, e, true, false, ""
|
|
case meshcatalogclient.EntryNotFound:
|
|
return fp, e, false, false, fmt.Sprintf("the catalog half holds no entry of %s's build %s", module, fp)
|
|
}
|
|
return fp, e, false, true, fmt.Sprintf("the entry of %s's build %s is asked of the catalog half and not yet answered", module, fp)
|
|
}
|
|
|
|
// askBuildsEntries asks the entry of each build in fps, the builds the node runs or its wants name, that the composer does not hold,
|
|
// and says the first not yet answered: the shadow composes a candidate from entries alone (§421, §628), and an answer wakes it. The
|
|
// fingerprints are kept beside the candidates' added ones while a proposal stands (addedFps).
|
|
func (c *Composer) askBuildsEntries(fps map[string]string) (pending string) {
|
|
if !c.byEntries() {
|
|
return ""
|
|
}
|
|
for _, module := range slices.Sorted(maps.Keys(fps)) {
|
|
fp := fps[module]
|
|
if !strings.HasPrefix(fp, meshcatalogclient.SourcePrefix2) {
|
|
continue
|
|
}
|
|
c.mu.Lock()
|
|
if c.addedFps == nil {
|
|
c.addedFps = map[string]bool{}
|
|
}
|
|
c.addedFps[fp] = true
|
|
c.mu.Unlock()
|
|
e, state := c.entryOf(fp)
|
|
switch {
|
|
case state == meshcatalogclient.EntryAnswered:
|
|
c.mu.Lock()
|
|
if _, kept := c.keptEnt[fp]; !kept && e.Module != "" {
|
|
c.keptEnt[fp] = keptEntry{entry: e, since: c.now()}
|
|
}
|
|
c.mu.Unlock()
|
|
case state == meshcatalogclient.EntryAsked && pending == "":
|
|
pending = fmt.Sprintf("the entry of %s's build %s is asked of the catalog half and not yet answered", module, fp)
|
|
}
|
|
}
|
|
return pending
|
|
}
|