Files
mesh-host/internal/selfcompose/entries.go
T

643 lines
23 KiB
Go

package selfcompose
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"git.novox.be/novox/mesh-sdk/go/buscalls"
"reflect"
"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 its entry and its parts alone (the builder's done key is not read): 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.
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)
}