sdk: the reference forms a build keeps and holds by, and the keeper of what each node's engine reads, shared by the assigner, the builder and the view (ADR 0618, ADR 0624, issue 774)
Logic only mesh-delivery's parts share, so it is mesh-delivery's and not mesh-sdk's (ADR 0593 §1, ADR 0608 §5). Tested through the assigner's and the builder's rows (the test budget). mesh-sdk is pinned by a local replace to the group's head while developing. Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
@@ -0,0 +1,495 @@
|
||||
// Package enginereads is how a mesh-delivery part learns what each node's engine reads of a manifest (novox/hq ADR 0618
|
||||
// §3, §4): it asks the node seat node-engine's verb manifest-reads, on the part's start, on the seat's event
|
||||
// manifest-reads-changed for that node and when a person asks again, and keeps each node's last answer in a state of its
|
||||
// own, one key per node, so a node that is away keeps the statement it last made (ADR 0624 §1). The assigner, the
|
||||
// builder and the delivery view share it; it is mesh-delivery's, not mesh-sdk's (ADR 0593 §1, ADR 0608 §5).
|
||||
package enginereads
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/busretry"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// The verb and the event as a manifest names them, and the subject the event is said on with the engine's machine last.
|
||||
const (
|
||||
Verb = "manifest-reads"
|
||||
Event = "manifest-reads-changed"
|
||||
Invoke = "seat:" + catalogue.EngineSeat + "." + Verb
|
||||
Consumes = catalogue.EngineSeat + "." + Event
|
||||
SubjectPrefix = "mesh.seat." + catalogue.EngineSeat + ".event." + Event + "."
|
||||
)
|
||||
|
||||
// Where a node's statement came from.
|
||||
const (
|
||||
// FromVerb: the node's engine answered the verb.
|
||||
FromVerb = "verb"
|
||||
// FromRecord: the engine did not answer it (an engine built before the verb, or nobody answered) and the node's record
|
||||
// `<node>.node` was read once in its place (the fallback, until every counted engine serves the verb; issue 774).
|
||||
FromRecord = "record"
|
||||
// FromNone: neither said anything, so the node is read as stating the baseline (ADR 0614 §4).
|
||||
FromNone = "none"
|
||||
)
|
||||
|
||||
// Version is the version of a statement this package writes and the only one it reads.
|
||||
const Version = 1
|
||||
|
||||
// Statement is one node's key: what its engine states it reads, when it stated it, when it was asked, where it came
|
||||
// from, whether the engine answered that ask (it is present), and, while a person's ask again waits, that ask.
|
||||
type Statement struct {
|
||||
V int `json:"v"`
|
||||
Node string `json:"node"`
|
||||
ManifestReads []string `json:"manifest-reads,omitempty"`
|
||||
StatedAt time.Time `json:"stated-at,omitzero"`
|
||||
AskedAt time.Time `json:"asked-at,omitzero"`
|
||||
From string `json:"from"`
|
||||
Answered bool `json:"answered"`
|
||||
Words string `json:"words,omitempty"`
|
||||
AskAgain *Again `json:"ask-again,omitempty"`
|
||||
}
|
||||
|
||||
// Again is a person's ask again of one node, written by the tool and answered by the part that keeps the state.
|
||||
type Again struct {
|
||||
By string `json:"by"`
|
||||
Why string `json:"why"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Encode is s as JSON at Version.
|
||||
func Encode(s Statement) ([]byte, error) {
|
||||
if s.Node == "" || strings.ContainsAny(s.Node, ".*> ") {
|
||||
return nil, fmt.Errorf("enginereads: %q is no node", s.Node)
|
||||
}
|
||||
s.V = Version
|
||||
return json.Marshal(s)
|
||||
}
|
||||
|
||||
// Decode reads a statement; it refuses another version and a value naming no node.
|
||||
func Decode(raw []byte) (Statement, error) {
|
||||
var s Statement
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return Statement{}, err
|
||||
}
|
||||
if s.V != Version {
|
||||
return Statement{}, fmt.Errorf("enginereads: version %d, not %d", s.V, Version)
|
||||
}
|
||||
if s.Node == "" {
|
||||
return Statement{}, errors.New("enginereads: a statement names no node")
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// Caller calls a seat's verb on one node: process.Conn.CallOn.
|
||||
type Caller func(node, key string, body any) (json.RawMessage, error)
|
||||
|
||||
// Ask asks node's engine the verb once and says what it answered. An engine that answered the call with an error is
|
||||
// present and states nothing through it; one that did not answer is not present.
|
||||
func Ask(call Caller, node string, now time.Time) Statement {
|
||||
s := Statement{Node: node, AskedAt: now.UTC(), From: FromNone}
|
||||
raw, err := call(node, Invoke, map[string]any{})
|
||||
var answered *busretry.Answered
|
||||
switch {
|
||||
case err == nil:
|
||||
a, derr := delivery.DecodeManifestReadsAnswer(raw)
|
||||
if derr != nil {
|
||||
s.Answered, s.Words = true, "the engine answered what is no statement: "+derr.Error()
|
||||
return s
|
||||
}
|
||||
s.ManifestReads, s.StatedAt, s.From, s.Answered = slices.Clone(a.ManifestReads), a.StatedAt.UTC(), FromVerb, true
|
||||
case errors.As(err, &answered):
|
||||
s.Answered, s.Words = true, "the engine does not serve "+Verb+": "+answered.Words
|
||||
default:
|
||||
s.Words = "nobody answered " + Verb + ": " + err.Error()
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// NodeOf is the node an event of manifest-reads-changed is about, read from its subject's last token and held equal to
|
||||
// its body's; ok false for a subject of another event.
|
||||
func NodeOf(subject string, data []byte) (string, bool, error) {
|
||||
node, ok := strings.CutPrefix(subject, SubjectPrefix)
|
||||
if !ok || node == "" || strings.Contains(node, ".") {
|
||||
return "", false, nil
|
||||
}
|
||||
e, err := delivery.DecodeManifestReadsChanged(data)
|
||||
if err != nil {
|
||||
return node, true, err
|
||||
}
|
||||
if e.Node != node {
|
||||
return node, true, fmt.Errorf("the event on %s names the node %s", subject, e.Node)
|
||||
}
|
||||
return node, true, nil
|
||||
}
|
||||
|
||||
// Keeper keeps every node's statement for one part: in memory, and in the part's state when Put is set, so a statement
|
||||
// outlives the part's restart. Nothing in it waits on a timer: a node is asked when the part says so.
|
||||
type Keeper struct {
|
||||
// Call asks the verb; nil asks nothing (every node then reads as the fallback or the baseline).
|
||||
Call Caller
|
||||
// Fallback reads a node's record `<node>.node` once where its engine did not answer the verb; nil reads none.
|
||||
Fallback func(node string) ([]string, bool)
|
||||
// Put writes a node's key of the part's state, Delete removes it; nil keeps the statements in memory only.
|
||||
Put func(node string, value []byte) error
|
||||
Delete func(node string) error
|
||||
// Changed is told after a node's statement changed.
|
||||
Changed func()
|
||||
Logf func(string, ...any)
|
||||
Now func() time.Time
|
||||
// Sync asks in the caller's goroutine (tests); otherwise each ask runs on its own.
|
||||
Sync bool
|
||||
|
||||
mu sync.Mutex
|
||||
by map[string]Statement
|
||||
asking map[string]bool
|
||||
}
|
||||
|
||||
func (k *Keeper) logf(f string, a ...any) {
|
||||
if k.Logf != nil {
|
||||
k.Logf(f, a...)
|
||||
}
|
||||
}
|
||||
|
||||
func (k *Keeper) now() time.Time {
|
||||
if k.Now != nil {
|
||||
return k.Now()
|
||||
}
|
||||
return time.Now()
|
||||
}
|
||||
|
||||
// AskNode asks node's engine, unless an ask of it runs already; why is said in the log.
|
||||
func (k *Keeper) AskNode(node, why string) {
|
||||
k.mu.Lock()
|
||||
if k.asking == nil {
|
||||
k.asking = map[string]bool{}
|
||||
}
|
||||
if k.asking[node] {
|
||||
k.mu.Unlock()
|
||||
return
|
||||
}
|
||||
k.asking[node] = true
|
||||
k.mu.Unlock()
|
||||
if k.Sync {
|
||||
k.ask(node, why)
|
||||
return
|
||||
}
|
||||
go k.ask(node, why)
|
||||
}
|
||||
|
||||
func (k *Keeper) ask(node, why string) {
|
||||
defer func() {
|
||||
k.mu.Lock()
|
||||
delete(k.asking, node)
|
||||
k.mu.Unlock()
|
||||
}()
|
||||
s := Statement{Node: node, AskedAt: k.now().UTC(), From: FromNone, Words: "nothing asks the verb"}
|
||||
if k.Call != nil {
|
||||
s = Ask(k.Call, node, k.now())
|
||||
}
|
||||
if s.From != FromVerb && k.Fallback != nil {
|
||||
if reads, ok := k.Fallback(node); ok {
|
||||
s.ManifestReads, s.From = slices.Clone(reads), FromRecord
|
||||
k.logf("%s on %s (%s): %s; its record %s.node is read once in its place (until every counted engine serves the verb)",
|
||||
Verb, node, why, s.Words, node)
|
||||
}
|
||||
}
|
||||
if s.From == FromNone {
|
||||
k.logf("%s on %s (%s): %s; the node reads as its last statement, or the baseline", Verb, node, why, s.Words)
|
||||
}
|
||||
k.take(s)
|
||||
}
|
||||
|
||||
// take keeps s, unless the node did not answer and an earlier statement stands: an away node keeps its last one (ADR 0624
|
||||
// §1), marked as not answering.
|
||||
func (k *Keeper) take(s Statement) {
|
||||
k.mu.Lock()
|
||||
if k.by == nil {
|
||||
k.by = map[string]Statement{}
|
||||
}
|
||||
if old, ok := k.by[s.Node]; ok && !s.Answered && old.From != FromNone {
|
||||
old.Answered, old.AskedAt, old.Words, old.AskAgain = false, s.AskedAt, s.Words, nil
|
||||
s = old
|
||||
}
|
||||
k.by[s.Node] = s
|
||||
k.mu.Unlock()
|
||||
if k.Put != nil {
|
||||
raw, err := Encode(s)
|
||||
if err == nil {
|
||||
err = k.Put(s.Node, raw)
|
||||
}
|
||||
if err != nil {
|
||||
k.logf("the statement of %s could not be kept: %v", s.Node, err)
|
||||
}
|
||||
}
|
||||
if k.Changed != nil {
|
||||
k.Changed()
|
||||
}
|
||||
}
|
||||
|
||||
// Heard takes a change of the part's own state in: at the start, every node's last statement; later, a person's ask again
|
||||
// (asked now) or a key a person dropped (forgotten). The part's own writes come back here and change nothing.
|
||||
func (k *Keeper) Heard(node string, raw []byte, deleted bool) {
|
||||
if deleted {
|
||||
k.mu.Lock()
|
||||
_, had := k.by[node]
|
||||
delete(k.by, node)
|
||||
k.mu.Unlock()
|
||||
if had && k.Changed != nil {
|
||||
k.Changed()
|
||||
}
|
||||
return
|
||||
}
|
||||
s, err := Decode(raw)
|
||||
if err != nil || s.Node != node {
|
||||
k.logf("the key %s of the engine reads is no statement of it, left: %v", node, err)
|
||||
return
|
||||
}
|
||||
if s.AskAgain != nil {
|
||||
k.AskNode(node, fmt.Sprintf("asked again by %s: %s", s.AskAgain.By, s.AskAgain.Why))
|
||||
return
|
||||
}
|
||||
k.mu.Lock()
|
||||
if k.by == nil {
|
||||
k.by = map[string]Statement{}
|
||||
}
|
||||
_, had := k.by[node]
|
||||
if !had {
|
||||
k.by[node] = s
|
||||
}
|
||||
k.mu.Unlock()
|
||||
if !had && k.Changed != nil {
|
||||
k.Changed()
|
||||
}
|
||||
}
|
||||
|
||||
// Forget drops a node the part no longer counts (it left the placement): its key too.
|
||||
func (k *Keeper) Forget(node string) {
|
||||
k.mu.Lock()
|
||||
_, had := k.by[node]
|
||||
delete(k.by, node)
|
||||
k.mu.Unlock()
|
||||
if had && k.Delete != nil {
|
||||
if err := k.Delete(node); err != nil {
|
||||
k.logf("the statement of %s could not be removed: %v", node, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Known says whether the part holds a statement of node, from any source.
|
||||
func (k *Keeper) Known(node string) bool {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
_, ok := k.by[node]
|
||||
return ok
|
||||
}
|
||||
|
||||
// Statements is every node's statement, by node.
|
||||
func (k *Keeper) Statements() map[string]Statement {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
out := make(map[string]Statement, len(k.by))
|
||||
for n, s := range k.by {
|
||||
out[n] = s
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Reads is, of the nodes given, what each states it reads: its last statement, nil (the baseline) for a node with none.
|
||||
func (k *Keeper) Reads(nodes []string) map[string][]string {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
out := make(map[string][]string, len(nodes))
|
||||
for _, n := range nodes {
|
||||
out[n] = slices.Clone(k.by[n].ManifestReads)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Present is, of the nodes given, those whose engine answered when last asked, with what each states.
|
||||
func (k *Keeper) Present(nodes []string) map[string][]string {
|
||||
k.mu.Lock()
|
||||
defer k.mu.Unlock()
|
||||
out := map[string][]string{}
|
||||
for _, n := range nodes {
|
||||
if s, ok := k.by[n]; ok && s.Answered {
|
||||
out[n] = slices.Clone(s.ManifestReads)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// State is what the tool needs of the part's state: the runtime's (stdio.KeptState through KeptState), or a test's.
|
||||
type State interface {
|
||||
Keys() ([]string, error)
|
||||
Get(key string) ([]byte, uint64, bool, error)
|
||||
Put(key string, value []byte) (uint64, error)
|
||||
Delete(key string) error
|
||||
// Await waits at most bound for a revision of key after after whose value done says is the answer.
|
||||
Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error)
|
||||
}
|
||||
|
||||
// Read lists every node's statement in st, by node name, or one node's; a key that does not read is named with why.
|
||||
func Read(st State, node string) (map[string]any, error) {
|
||||
keys := []string{node}
|
||||
if node == "" {
|
||||
var err error
|
||||
if keys, err = st.Keys(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
sort.Strings(keys)
|
||||
nodes := map[string]any{}
|
||||
for _, key := range keys {
|
||||
raw, _, found, err := st.Get(key)
|
||||
switch {
|
||||
case err != nil:
|
||||
return nil, err
|
||||
case !found:
|
||||
nodes[key] = map[string]any{"from": FromNone, "words": "no statement is kept: read as the baseline"}
|
||||
continue
|
||||
}
|
||||
s, derr := Decode(raw)
|
||||
if derr != nil {
|
||||
nodes[key] = map[string]any{"unread": derr.Error()}
|
||||
continue
|
||||
}
|
||||
nodes[key] = s
|
||||
}
|
||||
return map[string]any{"nodes": nodes}, nil
|
||||
}
|
||||
|
||||
// AskAgainOf writes node's key with a person's ask again and waits at most bound for the part's answer: the next
|
||||
// statement it writes there.
|
||||
func AskAgainOf(st State, node, by, why string, now time.Time, bound time.Duration) (Statement, error) {
|
||||
if node == "" || strings.ContainsAny(node, ".*> ") {
|
||||
return Statement{}, fmt.Errorf("%q is no node", node)
|
||||
}
|
||||
if strings.TrimSpace(why) == "" {
|
||||
return Statement{}, errors.New("say why it is asked again")
|
||||
}
|
||||
s := Statement{Node: node, From: FromNone}
|
||||
if raw, _, found, err := st.Get(node); err != nil {
|
||||
return Statement{}, err
|
||||
} else if found {
|
||||
if old, err := Decode(raw); err == nil {
|
||||
s = old
|
||||
}
|
||||
}
|
||||
s.AskAgain = &Again{By: by, Why: why, At: now.UTC()}
|
||||
raw, err := Encode(s)
|
||||
if err != nil {
|
||||
return Statement{}, err
|
||||
}
|
||||
rev, err := st.Put(node, raw)
|
||||
if err != nil {
|
||||
return Statement{}, err
|
||||
}
|
||||
answer, err := st.Await(node, rev, func(v []byte) bool {
|
||||
got, err := Decode(v)
|
||||
return err == nil && got.AskAgain == nil
|
||||
}, bound)
|
||||
if err != nil {
|
||||
return Statement{}, fmt.Errorf("the part did not answer the ask again of %s within %s: %w", node, bound, err)
|
||||
}
|
||||
return Decode(answer)
|
||||
}
|
||||
|
||||
// KeptState is a part's state as the runtime keeps it, for the tool.
|
||||
func KeptState(s stdio.KeptState) State { return keptState{s} }
|
||||
|
||||
type keptState struct{ s stdio.KeptState }
|
||||
|
||||
func (k keptState) Keys() ([]string, error) { return k.s.Keys() }
|
||||
|
||||
func (k keptState) Get(key string) ([]byte, uint64, bool, error) {
|
||||
e, err := k.s.Get(key)
|
||||
if err != nil || e == nil {
|
||||
return nil, 0, false, err
|
||||
}
|
||||
return e.Value, e.Revision, true, nil
|
||||
}
|
||||
|
||||
func (k keptState) Put(key string, value []byte) (uint64, error) { return k.s.Put(key, json.RawMessage(value)) }
|
||||
|
||||
func (k keptState) Delete(key string) error { return k.s.Delete(key) }
|
||||
|
||||
// Await watches key for the answer, and reads it once too in case the watch heard it before it began; a context's bound,
|
||||
// no timer (ADR 0437 §8).
|
||||
func (k keptState) Await(key string, after uint64, done func([]byte) bool, bound time.Duration) ([]byte, error) {
|
||||
heard := make(chan []byte, 1)
|
||||
if err := k.s.Watch(key, func(c stdio.StateChange) error {
|
||||
if c.Op == "put" && c.Revision > after && done(c.Value) {
|
||||
select {
|
||||
case heard <- c.Value:
|
||||
default:
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if e, err := k.s.Get(key); err == nil && e != nil && e.Revision > after && done(e.Value) {
|
||||
return e.Value, nil
|
||||
}
|
||||
wait, cancel := context.WithTimeout(context.Background(), bound)
|
||||
defer cancel()
|
||||
select {
|
||||
case v := <-heard:
|
||||
return v, nil
|
||||
case <-wait.Done():
|
||||
return nil, errors.New("it is not running, or cannot read the verb's answer")
|
||||
}
|
||||
}
|
||||
|
||||
// AnswerWithin is how long the tool waits for the part to answer an ask again: the verb's own call bound and a margin.
|
||||
var AnswerWithin = 25 * time.Second
|
||||
|
||||
// Tool is a part's tool over its state of engine reads (tools-first, ADR 0618 §3): read every node's statement or one,
|
||||
// ask one node's engine again now, and, where drops, drop one node's key (a node the part keeps by mistake). owner names
|
||||
// the part in the words.
|
||||
func Tool(name, owner string, st State, drops bool, now func() time.Time) stdio.Tool {
|
||||
props := map[string]any{
|
||||
"node": map[string]any{"type": "string", "description": "one node; every node when left out"},
|
||||
"ask_again": map[string]any{"type": "string", "description": "\"true\": ask the node's engine " + Verb + " again now, and answer what it states"},
|
||||
"why": map[string]any{"type": "string", "description": "why it is asked again or dropped, in plain words"},
|
||||
}
|
||||
description := "What each node's engine states it reads of a manifest (manifest words, forms and seat names, novox/hq ADR 0618 §3), " +
|
||||
"as " + owner + " last asked the node-engine seat's verb " + Verb + " and uses it: per node its capabilities, when the engine " +
|
||||
"stated them and when it was asked, whether it answered (present) and where the statement came from: verb, record (the " +
|
||||
"engine did not answer and its node record was read once in its place) or none (read as the baseline). A node that is " +
|
||||
"away keeps its last statement. With node and ask_again, " + owner + " asks that node's engine again and answers its new " +
|
||||
"statement."
|
||||
if drops {
|
||||
props["drop"] = map[string]any{"type": "string", "description": "\"true\": drop the node's key, so it reads as stating the baseline until its engine is asked again"}
|
||||
description += " With node and drop, the node's key is dropped."
|
||||
}
|
||||
return stdio.Tool{Name: name, Description: description, Input: map[string]any{"type": "object", "properties": props},
|
||||
Run: func(args map[string]any) (any, error) {
|
||||
str := func(k string) string { s, _ := args[k].(string); return strings.TrimSpace(s) }
|
||||
node := str("node")
|
||||
switch {
|
||||
case str("ask_again") == "true":
|
||||
return AskAgainOf(st, node, stdio.CallerOf(args), str("why"), now(), AnswerWithin)
|
||||
case drops && str("drop") == "true":
|
||||
if node == "" || str("why") == "" {
|
||||
return nil, errors.New("drop takes node and why")
|
||||
}
|
||||
if err := st.Delete(node); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return map[string]any{"node": node, "dropped": true,
|
||||
"next": "the node reads as stating the baseline until its engine is asked again (ask_again, its event, or " + owner + "'s start)"}, nil
|
||||
}
|
||||
return Read(st, node)
|
||||
}}
|
||||
}
|
||||
@@ -0,0 +1,226 @@
|
||||
// Package forms holds the artifact reference forms the builder writes a done build key in and the assigner holds a want
|
||||
// by (novox/hq ADR 0618 §1-§2, ADR 0624): logic only those two modules of mesh-delivery share, so it is mesh-delivery's
|
||||
// and not mesh-sdk's (ADR 0593 §1, ADR 0608 §5). The build key's fields (delivery.BuilderBuild.Needs,
|
||||
// delivery.BuiltArtifact.References) are the contract, in mesh-sdk.
|
||||
package forms
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// artifactStore opens the artifact store's references (delivery.BuiltArtifact).
|
||||
const artifactStore = "artifact-store://"
|
||||
|
||||
// The artifact reference forms (novox/hq ADR 0605 §9, ADR 0618 §1-§2, ADR 0624): each form a done key's reference may be
|
||||
// written in, named as the reader capability `form:<name>` a node-engine states when it reads it. The forms written
|
||||
// before any engine stated what it reads are first forms: every engine reads them, so none needs a statement, and a
|
||||
// node whose engine states nothing (the baseline, ADR 0614 §4) reads exactly those. A new form is added to
|
||||
// Current, newest last, in the change that teaches the node-engine to read it.
|
||||
|
||||
// The first forms, by name.
|
||||
const (
|
||||
// FormArtifactStore: artifact-store://<module>/<artifact>/blobs/<digest>, or an image's kept artifact-store://…@<digest>.
|
||||
FormArtifactStore = "artifact-store"
|
||||
// FormRegistry: an image's or an upstream copy's <registry>/<module>/<artifact>@<digest>.
|
||||
FormRegistry = "registry"
|
||||
// FormPackageVersion: a package's <name>@<version>.
|
||||
FormPackageVersion = "package-version"
|
||||
)
|
||||
|
||||
// ArtifactForm is one reference form: its name, whether it is a first form, the artifact kinds it names, and how a
|
||||
// reference is told to be in it.
|
||||
type ArtifactForm struct {
|
||||
Name string
|
||||
First bool
|
||||
Kinds []string
|
||||
Is func(reference string) bool
|
||||
}
|
||||
|
||||
// ArtifactForms are the forms a builder knows, oldest first.
|
||||
type ArtifactForms []ArtifactForm
|
||||
|
||||
// Current are the forms the builder writes and the assigner knows: the three first forms (delivery.BuiltArtifact's
|
||||
// reference by kind).
|
||||
var Current = ArtifactForms{
|
||||
{Name: FormArtifactStore, First: true, Is: func(r string) bool { return strings.HasPrefix(r, artifactStore) },
|
||||
Kinds: []string{catalogue.ArtifactBundle, catalogue.ArtifactArchive, catalogue.ArtifactImage, catalogue.ArtifactUpstream}},
|
||||
{Name: FormRegistry, First: true, Is: func(r string) bool { return !strings.HasPrefix(r, artifactStore) && strings.Contains(r, "@sha256:") },
|
||||
Kinds: []string{catalogue.ArtifactImage, catalogue.ArtifactUpstream}},
|
||||
{Name: FormPackageVersion, First: true, Is: func(r string) bool { return strings.LastIndex(r, "@") > 0 && !strings.Contains(r, "@sha256:") },
|
||||
Kinds: []string{catalogue.ArtifactPackage}},
|
||||
}
|
||||
|
||||
// FormNeed is the reader capability a form needs of a node-engine: `form:<name>`.
|
||||
func FormNeed(form string) string { return catalogue.ReadForm + form }
|
||||
|
||||
// FormOf is the name of the form reference is written in for an artifact of kind, and whether one of fs names it.
|
||||
func (fs ArtifactForms) FormOf(kind, reference string) (string, bool) {
|
||||
for i := len(fs) - 1; i >= 0; i-- { // the newest first: a newer form may be a narrower shape of an older one
|
||||
f := fs[i]
|
||||
if (kind == "" || slices.Contains(f.Kinds, kind)) && f.Is(reference) {
|
||||
return f.Name, true
|
||||
}
|
||||
}
|
||||
return "", false
|
||||
}
|
||||
|
||||
// Reads says whether an engine whose statement is stated reads form: a first form always (the baseline included), any
|
||||
// other only when stated names `form:<name>`. A form fs does not know is read only when stated.
|
||||
func (fs ArtifactForms) Reads(stated []string, form string) bool {
|
||||
for _, f := range fs {
|
||||
if f.Name == form && f.First {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return slices.Contains(stated, FormNeed(form))
|
||||
}
|
||||
|
||||
// Needs are the reader capabilities a done build's artifacts carry (ADR 0618 §1): `form:<name>` of every form any of
|
||||
// its references, single or listed, is written in, sorted. Nil when it names none (no artifact, or no reference).
|
||||
func (fs ArtifactForms) Needs(artifacts []delivery.BuiltArtifact) []string {
|
||||
set := map[string]bool{}
|
||||
for _, a := range artifacts {
|
||||
for _, r := range append([]string{a.Reference}, a.References...) {
|
||||
if r == "" {
|
||||
continue
|
||||
}
|
||||
if form, ok := fs.FormOf(a.Kind, r); ok {
|
||||
set[FormNeed(form)] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if len(set) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make([]string, 0, len(set))
|
||||
for n := range set {
|
||||
out = append(out, n)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// Held says whether a build's reference forms hold a node's want (ADR 0624 §2): only when the build names forms and the
|
||||
// node's engine, its statement stated, reads none of them. A build that names none holds no want.
|
||||
func (fs ArtifactForms) Held(needs, stated []string) bool {
|
||||
if len(needs) == 0 {
|
||||
return false
|
||||
}
|
||||
for _, n := range needs {
|
||||
if form, ok := strings.CutPrefix(n, catalogue.ReadForm); ok && fs.Reads(stated, form) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// Writable are the forms of kind every present node's engine reads (ADR 0618 §2: a new form is written only once every
|
||||
// present engine states it), oldest first, by node its statement. With no node present, the first forms only.
|
||||
func (fs ArtifactForms) Writable(kind string, present map[string][]string) []string {
|
||||
var out []string
|
||||
for _, f := range fs {
|
||||
if !slices.Contains(f.Kinds, kind) {
|
||||
continue
|
||||
}
|
||||
every := f.First || len(present) > 0
|
||||
for _, stated := range present {
|
||||
every = every && fs.Reads(stated, f.Name)
|
||||
}
|
||||
if every {
|
||||
out = append(out, f.Name)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Kept are the forms a build key names an artifact of kind in (ADR 0624 §1, §3), oldest first, so the first is the one
|
||||
// the single reference holds: the newest form every present engine reads (Writable), then, for each counted node, present
|
||||
// or away, whose engine reads none of those kept so far, the newest form of kind it reads. A node stating nothing reads
|
||||
// the first forms. A form leaves only once no counted node reads it alone; a node leaves the count when it is retired.
|
||||
func (fs ArtifactForms) Kept(kind string, present, counted map[string][]string) []string {
|
||||
writable := fs.Writable(kind, present)
|
||||
kept := map[string]bool{}
|
||||
if len(writable) > 0 {
|
||||
kept[writable[len(writable)-1]] = true
|
||||
}
|
||||
nodes := make([]string, 0, len(counted))
|
||||
for node := range counted {
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
for _, node := range nodes {
|
||||
stated := counted[node]
|
||||
var newest string
|
||||
readsKept := false
|
||||
for _, f := range fs {
|
||||
if !slices.Contains(f.Kinds, kind) || !fs.Reads(stated, f.Name) {
|
||||
continue
|
||||
}
|
||||
newest = f.Name
|
||||
readsKept = readsKept || kept[f.Name]
|
||||
}
|
||||
if !readsKept && newest != "" {
|
||||
kept[newest] = true
|
||||
}
|
||||
}
|
||||
var out []string
|
||||
for _, f := range fs {
|
||||
if kept[f.Name] {
|
||||
out = append(out, f.Name)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Choose names a built artifact by the references the build answered in the forms Kept keeps (ADR 0624 §1, §3): the
|
||||
// oldest kept form in the single Reference, every kept one in References when there are more than one. A reference in a
|
||||
// form Kept does not keep is left out, unless none of the answered is kept: then the artifact keeps what was answered,
|
||||
// its first answered in Reference, since a key names where its artifact is (ADR 0498 clause 1).
|
||||
func (fs ArtifactForms) Choose(a delivery.BuiltArtifact, answered []string, present, counted map[string][]string) delivery.BuiltArtifact {
|
||||
kept := fs.Kept(a.Kind, present, counted)
|
||||
type ref struct {
|
||||
at int
|
||||
reference string
|
||||
}
|
||||
var in []ref
|
||||
for _, r := range answered {
|
||||
if r == "" {
|
||||
continue
|
||||
}
|
||||
form, ok := fs.FormOf(a.Kind, r)
|
||||
if at := slices.Index(kept, form); ok && at >= 0 {
|
||||
in = append(in, ref{at, r})
|
||||
}
|
||||
}
|
||||
sort.SliceStable(in, func(i, j int) bool { return in[i].at < in[j].at })
|
||||
a.References = nil
|
||||
switch {
|
||||
case len(in) > 0:
|
||||
a.Reference = in[0].reference
|
||||
if len(in) > 1 {
|
||||
for _, r := range in {
|
||||
a.References = append(a.References, r.reference)
|
||||
}
|
||||
}
|
||||
default:
|
||||
a.Reference = ""
|
||||
for _, r := range answered {
|
||||
if r == "" {
|
||||
continue
|
||||
}
|
||||
if a.Reference == "" {
|
||||
a.Reference = r
|
||||
}
|
||||
a.References = append(a.References, r)
|
||||
}
|
||||
if len(a.References) < 2 {
|
||||
a.References = nil
|
||||
}
|
||||
}
|
||||
return a
|
||||
}
|
||||
+11
@@ -3,3 +3,14 @@ module git.novox.be/novox/mesh-delivery/sdk
|
||||
go 1.26.0
|
||||
|
||||
require git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8
|
||||
|
||||
require (
|
||||
github.com/klauspost/compress v1.20.0 // indirect
|
||||
github.com/nats-io/nats.go v1.54.0 // indirect
|
||||
github.com/nats-io/nkeys v0.4.16 // indirect
|
||||
github.com/nats-io/nuid v1.0.1 // indirect
|
||||
golang.org/x/crypto v0.57.0 // indirect
|
||||
golang.org/x/sys v0.48.0 // indirect
|
||||
)
|
||||
|
||||
replace git.novox.be/novox/mesh-sdk/go => ../../0618-sdk/go
|
||||
|
||||
@@ -1,5 +1,3 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8 h1:XEbIVImp6yGjA91rp0iZBoDdL5KfamjOuDLwWvqh0hc=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261103221822-85ba0a9a70d8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/klauspost/compress v1.20.0 h1:a3C1ke2ohxFymNlb2HWAHjDeKCI90scRskErZkR0ezA=
|
||||
github.com/klauspost/compress v1.20.0/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
|
||||
github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
|
||||
|
||||
Reference in New Issue
Block a user