novox/mesh-host
18
0
Fork 0
Code Issues Pull Requests 1 Releases Wiki Activity

Merge branch 'b6/registry' into feat/registry-milestone

Browse Source
This commit is contained in:
jschoubben
2026-11-05 15:23:24 +01:00
parent ec3a65fa8d 7eb7b9fa62
commit 3cf4fbaf3d
44 changed files with 650 additions and 549 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/link/bus.go
+4 -5
View File
@@ -4,10 +4,9 @@ import (
"context"
"errors"
"fmt"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-host/internal/buckets"
"time"
)
// Bus is the link open now as the session holds it. What it can do is asked of it by what it implements (HaveBus,
@@ -87,9 +86,9 @@ func (b OverNATS) HaveBind(ctx context.Context) (HaveBucket, error) {
}
js = bounded
}
kv, err := js.KeyValue(delivery.HaveBucket)
kv, err := js.KeyValue(buckets.Have)
if err != nil {
return nil, fmt.Errorf("the have bucket %s: %w", delivery.HaveBucket, err)
return nil, fmt.Errorf("the have bucket %s: %w", buckets.Have, err)
}
return natsHaveBucket{kv}, nil
}
internal/link/enginedata.go
+3 -4
View File
@@ -4,17 +4,16 @@ import (
"context"
"errors"
"fmt"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-host/internal/buckets"
)
// ---- the data items of the modules placed on this node, in the bus's own storage (novox/hq ADR 0577 clause 1) ------
// DataBucket is the node-engine's declared state data (delivery.EngineDataBucket), created by the grants part from the
// DataBucket is the node-engine's declared state data (buckets.Data), created by the grants part from the
// engine's manifest as its states acts and health are: one key `<node>.<module>` per module placed on the node that
// declares data items, written by that node's engine alone, read by that node's disk-load.
const DataBucket = delivery.EngineDataBucket
const DataBucket = buckets.Data
// dataStore binds DataBucket over the link open now: bound, never created.
func (l *natsLink) dataStore() (nats.KeyValue, error) {
internal/link/have.go
+81 -93
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
"github.com/novox/mesh-host/internal/compositions"
@@ -12,8 +13,6 @@ import (
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// ErrHaveAbsent is what HaveGet returns for a key the bucket does not hold.
@@ -40,14 +39,14 @@ type HaveBucket interface {
type haveState struct {
mu sync.Mutex
modules map[string]ModuleFact
health map[string]delivery.Health
health map[string]string
// words is, by module, what the last look found of the resource that made its health failed or waits: the have says
// it, so a failed health is never without words (issue 1496: "the node gave no words" hid a directory used as found).
words map[string]string
// looked says a look's health was taken: a running module it judges no resource of then has nothing that could be
// down, and is ok (issue 1945).
looked bool
node *delivery.HaveNode
node *nodeengine.HaveByNodeNodeValue
// nodeComposed is the digest of the node's own composition it runs (ADR 0539 §9), kept beside the node's facts.
nodeComposed string
// nodeSealed is, on a node composing its own, each value its own composition carries sealed to its key (ADR 0547 §3,
@@ -65,7 +64,7 @@ type haveState struct {
removed map[string]string
// userList is what the bus node's engine last wrote of the bus's user list, by the module whose have says it (novox/hq
// ADR 0547 §8); kept apart from the module's fact, which every apply replaces.
userList map[string]delivery.UserListWritten
userList map[string]nodeengine.UserListWritten
// userListNoKey is each module whose user list was said to have no have key to ride on, so it is said once.
userListNoKey map[string]bool
// unjudged is each module whose composition moved (an apply began, or the fact names another build) and that no look
@@ -76,8 +75,8 @@ type haveState struct {
// inMotion says a module's phase is one of a change still under way: what a look finds of it then judges nothing of the
// composition being applied (a container stopped to be replaced reads unhealthy), so its health is not yet judged.
func inMotion(phase string) bool {
switch delivery.Phase(phase) {
case delivery.PhaseBackingUp, delivery.PhaseApplying, delivery.PhaseChecking:
switch string(phase) {
case nodeengineclient.PhaseBackingUp, nodeengineclient.PhaseApplying, nodeengineclient.PhaseChecking:
return true
}
return false
@@ -85,15 +84,15 @@ func inMotion(phase string) bool {
// healthOf is the health a module's have carries: unknown (not yet judged) while its change is under way or until a look
// after it, the last look's otherwise. The caller holds the lock.
func (s *haveState) healthOf(module string, f ModuleFact) delivery.Health {
func (s *haveState) healthOf(module string, f ModuleFact) string {
if inMotion(f.Phase) {
return delivery.HealthUnknown
return nodeengineclient.HealthUnknown
}
if got, known := s.health[module]; known {
return got
}
if s.unjudged[module] {
return delivery.HealthUnknown
return nodeengineclient.HealthUnknown
}
return nothingJudged(f, s.looked)
}
@@ -141,8 +140,8 @@ func (s *haveState) takeLook(h *Health) {
// wordsOf is what a have says of its health when the apply said nothing: the words of the resource that made it failed
// or waits, while it is so. The caller holds the lock.
func (s *haveState) wordsOf(module string, health delivery.Health) string {
if health != delivery.HealthFailed && health != delivery.HealthWaits {
func (s *haveState) wordsOf(module string, health string) string {
if health != nodeengineclient.HealthFailed && health != nodeengineclient.HealthWaits {
return ""
}
return s.words[module]
@@ -154,8 +153,8 @@ func (s *haveState) wordsOf(module string, health delivery.Health) string {
// module's health: it says nothing of how the module runs, waits for the operator's hand-over and keeps its own
// condition, raised from the same statement (issue 1496: on ace mesh-build-node's cache directory, there before the
// mesh, made its have failed while its check answered healthy, and the assigner held its version on every node).
func moduleHealths(h *Health) (map[string]delivery.Health, map[string]string) {
out, words := map[string]delivery.Health{}, map[string]string{}
func moduleHealths(h *Health) (map[string]string, map[string]string) {
out, words := map[string]string{}, map[string]string{}
if h == nil {
return out, words
}
@@ -183,7 +182,7 @@ func moduleHealths(h *Health) (map[string]delivery.Health, map[string]string) {
}
}
for module, r := range rank {
out[module] = [...]delivery.Health{delivery.HealthOK, delivery.HealthUnknown, delivery.HealthWaits, delivery.HealthFailed}[r]
out[module] = [...]string{nodeengineclient.HealthOK, nodeengineclient.HealthUnknown, nodeengineclient.HealthWaits, nodeengineclient.HealthFailed}[r]
if r < 2 {
delete(words, module)
}
@@ -206,21 +205,28 @@ func resourceWords(r ResourceHealth) string {
}
// nodeFacts is the node's own facts of a report, each as the JSON the report carries it in.
func nodeFacts(r Report) delivery.HaveNode {
n := delivery.HaveNode{V: sdkkeys.Version}
carry := func(into *json.RawMessage, empty bool, v any) {
func nodeFacts(r Report) nodeengine.HaveByNodeNodeValue {
n := nodeengine.HaveByNodeNodeValue{V: sdkkeys.Version}
carry := func(into *[]map[string]any, empty bool, v any) {
if empty {
return
}
if raw, err := json.Marshal(v); err == nil {
*into = raw
raw, err := json.Marshal(v)
if err != nil {
return
}
var list []map[string]any
if json.Unmarshal(raw, &list) == nil {
*into = list
}
}
carry(&n.Windows, len(r.Windows) == 0, r.Windows)
carry(&n.Busy, len(r.Busy) == 0, r.Busy)
carry(&n.Strays, len(r.Strays) == 0, r.Strays)
carry(&n.Filters, len(r.Filters) == 0, r.Filters)
carry(&n.Profile, len(r.Profile) == 0, r.Profile)
if len(r.Profile) > 0 {
n.Profile = r.Profile
}
return n
}
@@ -296,7 +302,7 @@ func (q *Session) noteApply(r Report) bool {
// nodeFactsWithRecord is the node's own facts of a report with its record (ADR 0555 §3): a report that says nothing of
// the machine leaves the record last made standing, and one without a health statement keeps the contract last stated.
func (q *Session) nodeFactsWithRecord(r Report, before *delivery.HaveNode) *delivery.HaveNode {
func (q *Session) nodeFactsWithRecord(r Report, before *nodeengine.HaveByNodeNodeValue) *nodeengine.HaveByNodeNodeValue {
n := nodeFacts(r)
if q.Keys != nil {
keys := q.Keys()
@@ -377,20 +383,20 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
sort.Strings(modules)
for _, module := range modules {
f := s.modules[module]
h := delivery.Have{V: sdkkeys.Version, FP: f.FP, Composed: f.Composed, Build: f.Build,
Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown, Detail: f.Detail}
h := nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Fp: f.FP, Composed: f.Composed, Build: f.Build,
Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthUnknown, Detail: f.Detail}
h.Kept = f.Kept
if delivery.Phase(f.Phase) == delivery.PhaseWaitingYes {
h.Judgement, h.Waits, h.BuildAlone = delivery.Judgement(f.Judgement), f.Waits, f.BuildAlone
if string(f.Phase) == nodeengineclient.PhaseWaitingYes {
h.Judgement, h.Waits, h.BuildAlone = string(f.Judgement), f.Waits, f.BuildAlone
}
switch {
case f.Phase != "":
h.Phase = delivery.Phase(f.Phase)
h.Phase = string(f.Phase)
case f.Failed:
h.Phase = delivery.PhaseFailed
h.Phase = nodeengineclient.PhaseFailed
}
h.SealedAt = f.Sealed
if h.Phase == delivery.PhaseFailed {
if h.Phase == nodeengineclient.PhaseFailed {
h.PullWait = f.PullWait
}
h.Health = s.healthOf(module, f)
@@ -400,19 +406,19 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
if w, ok := s.userList[module]; ok {
h.UserList = &w
}
key, err := delivery.HaveKey(node, module)
key, err := nodeengineclient.HaveKey(node, module)
if err != nil {
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
continue
}
raw, err := delivery.EncodeHave(h)
raw, err := nodeengineclient.EncodeHave(h)
if err != nil {
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
continue
}
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
c, err := delivery.DecodeHave(current)
return err == nil && delivery.SameHave(c, h)
c, err := nodeengineclient.DecodeHave(current)
return err == nil && nodeengineclient.SameHave(c, h)
}) {
return
}
@@ -428,19 +434,19 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) {
return
}
if s.node != nil {
key, err := delivery.HaveNodeKey(node)
key, err := nodeengineclient.HaveNodeKey(node)
if err == nil {
var raw []byte
facts := *s.node
facts.Composed = s.nodeComposed
facts.SealedAt = s.nodeSealed
facts.Candidates = interimAnswers(s.nodeCandidates)
facts.Candidates = s.nodeCandidates
facts.Composes, facts.ComposesWhy = s.nodeComposes, s.nodeComposesWhy
if raw, err = delivery.EncodeHaveNode(facts); err == nil {
if raw, err = nodeengineclient.EncodeHaveNode(facts); err == nil {
want := facts
if !q.putHave(write, bucket, key, raw, func(current []byte) bool {
c, err := delivery.DecodeHaveNode(current)
return err == nil && delivery.SameHaveNode(c, want)
c, err := nodeengineclient.DecodeHaveNode(current)
return err == nil && nodeengineclient.SameHaveNode(c, want)
}) {
return
}
@@ -503,15 +509,15 @@ func (q *Session) writeRemoved(ctx context.Context, b HaveBucket, node string) b
}
sort.Strings(gone)
for _, module := range gone {
key, err := delivery.HaveKey(node, module)
key, err := nodeengineclient.HaveKey(node, module)
if err != nil {
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
delete(s.removed, module)
continue
}
h := delivery.Have{V: sdkkeys.Version, Build: s.removed[module], Phase: delivery.PhaseRemoved,
Health: delivery.HealthUnknown, Detail: "the applied declaration no longer holds the module"}
raw, err := delivery.EncodeHave(h)
h := nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: s.removed[module], Phase: nodeengineclient.PhaseRemoved,
Health: nodeengineclient.HealthUnknown, Detail: "the applied declaration no longer holds the module"}
raw, err := nodeengineclient.EncodeHave(h)
if err != nil {
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
delete(s.removed, module)
@@ -531,7 +537,7 @@ func (q *Session) writeRemoved(ctx context.Context, b HaveBucket, node string) b
q.haveFailed(err)
return false
}
if c, err := delivery.DecodeHave(current); err == nil && c.Phase == delivery.PhaseRemoved {
if c, err := nodeengineclient.DecodeHave(current); err == nil && c.Phase == nodeengineclient.PhaseRemoved {
delete(s.removed, module)
continue
}
@@ -565,7 +571,7 @@ func (q *Session) writeUserListsAlone(ctx context.Context, b HaveBucket, node st
}
sort.Strings(alone)
for _, module := range alone {
key, err := delivery.HaveKey(node, module)
key, err := nodeengineclient.HaveKey(node, module)
if err != nil {
q.sayHave(fmt.Sprintf("module %q has no key in the have bucket: %v", module, err))
continue
@@ -592,7 +598,7 @@ func (q *Session) writeUserListsAlone(ctx context.Context, b HaveBucket, node st
return false
}
delete(s.userListNoKey, module)
h, err := delivery.DecodeHave(current)
h, err := nodeengineclient.DecodeHave(current)
if err != nil {
q.sayHave(fmt.Sprintf("the bus's user list is not written in %s: its value cannot be read: %v", key, err))
continue
@@ -600,10 +606,10 @@ func (q *Session) writeUserListsAlone(ctx context.Context, b HaveBucket, node st
w := s.userList[module]
next := h
next.UserList = &w
if delivery.SameHave(h, next) {
if nodeengineclient.SameHave(h, next) {
continue
}
raw, err := delivery.EncodeHave(next)
raw, err := nodeengineclient.EncodeHave(next)
if err != nil {
q.sayHave(fmt.Sprintf("cannot encode %s: %v", key, err))
continue
@@ -685,12 +691,12 @@ func (q *Session) NoteModule(ctx context.Context, f ModuleFact) {
// NoteUserList takes what the bus node's engine wrote of the bus's user list, and writes it in the bus module's have
// where it differs from the bucket's (novox/hq ADR 0547 §8). A module this engine has said no have of yet carries it
// from its first have on.
func (q *Session) NoteUserList(ctx context.Context, module string, w delivery.UserListWritten) {
func (q *Session) NoteUserList(ctx context.Context, module string, w nodeengine.UserListWritten) {
q.init()
s := &q.haves
s.mu.Lock()
if s.userList == nil {
s.userList = map[string]delivery.UserListWritten{}
s.userList = map[string]nodeengine.UserListWritten{}
}
s.userList[module] = w
s.mu.Unlock()
@@ -732,7 +738,7 @@ func (q *Session) NoteNode(ctx context.Context, composed string, sealed composit
s.nodeComposed = composed
s.nodeSealed = sealed
if s.node == nil {
n := delivery.HaveNode{V: sdkkeys.Version}
n := nodeengine.HaveByNodeNodeValue{V: sdkkeys.Version}
s.node = &n
}
s.mu.Unlock()
@@ -765,7 +771,7 @@ func (q *Session) NoteCandidates(ctx context.Context, answers map[string]nodeeng
q.writeHaves(ctx, bus)
}
// NoteComposes says, in the node's own key, whose compositions the node applies (delivery.ComposesOwn) and why: written
// NoteComposes says, in the node's own key, whose compositions the node applies (nodeengineclient.ComposesOwn) and why: written
// at each start of its agent, where every have is read (issue 1126).
func (q *Session) NoteComposes(ctx context.Context, composes, why string) {
q.init()
@@ -786,49 +792,49 @@ func (q *Session) NoteComposes(ctx context.Context, composes, why string) {
// ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition
// it last ran is and carries sealed, read after a restart (ADR 0547 §3).
func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) {
func (q *Session) ReadNodeHave(ctx context.Context) (nodeengine.HaveByNodeNodeValue, bool) {
n, ok, _ := q.ReadNodeHaveErr(ctx)
return n, ok
}
// ReadNodeHaveErr is ReadNodeHave telling the key absent (false, nil) from one that could not be read (false, the
// error): a bus that did not answer is never kept as a node that ran nothing (issue 1126, 2026-10-28).
func (q *Session) ReadNodeHaveErr(ctx context.Context) (delivery.HaveNode, bool, error) {
func (q *Session) ReadNodeHaveErr(ctx context.Context) (nodeengine.HaveByNodeNodeValue, bool, error) {
q.init()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
hb, ok := bus.(HaveBus)
if !ok || bus == nil {
return delivery.HaveNode{}, false, ErrNotLinkedHaves
return nodeengine.HaveByNodeNodeValue{}, false, ErrNotLinkedHaves
}
key, err := delivery.HaveNodeKey(q.Membership.Node)
key, err := nodeengineclient.HaveNodeKey(q.Membership.Node)
if err != nil {
return delivery.HaveNode{}, false, err
return nodeengine.HaveByNodeNodeValue{}, false, err
}
read, cancel := context.WithTimeout(ctx, q.timeout())
defer cancel()
bucket, err := hb.HaveBind(read)
if err != nil {
return delivery.HaveNode{}, false, err
return nodeengine.HaveByNodeNodeValue{}, false, err
}
value, _, err := bucket.Get(read, key)
if errors.Is(err, ErrHaveAbsent) {
return delivery.HaveNode{}, false, nil
return nodeengine.HaveByNodeNodeValue{}, false, nil
}
if err != nil {
return delivery.HaveNode{}, false, err
return nodeengine.HaveByNodeNodeValue{}, false, err
}
n, err := delivery.DecodeHaveNode(value)
n, err := nodeengineclient.DecodeHaveNode(value)
if err != nil {
return delivery.HaveNode{}, false, nil // a key no engine can read names nothing it runs
return nodeengine.HaveByNodeNodeValue{}, false, nil // a key no engine can read names nothing it runs
}
return n, true, nil
}
// ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot
// be read answers as none: the agent then compares with the declaration it kept, and applies only what differs.
func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) {
func (q *Session) ReadHave(ctx context.Context, module string) (nodeengine.HaveByNodeModuleValue, bool) {
h, ok, _ := q.ReadHaveErr(ctx, module)
return h, ok
}
@@ -838,62 +844,44 @@ var ErrNotLinkedHaves = errors.New("no link to the bus that binds the have bucke
// ReadHaveErr is ReadHave telling a key absent (false, nil) from one that could not be read (false, the error): a bus
// that does not answer for a moment is not a module that never ran (issue 1126).
func (q *Session) ReadHaveErr(ctx context.Context, module string) (delivery.Have, bool, error) {
func (q *Session) ReadHaveErr(ctx context.Context, module string) (nodeengine.HaveByNodeModuleValue, bool, error) {
q.init()
q.mu.Lock()
bus := q.bus
q.mu.Unlock()
hb, ok := bus.(HaveBus)
if !ok || bus == nil {
return delivery.Have{}, false, ErrNotLinkedHaves
return nodeengine.HaveByNodeModuleValue{}, false, ErrNotLinkedHaves
}
key, err := delivery.HaveKey(q.Membership.Node, module)
key, err := nodeengineclient.HaveKey(q.Membership.Node, module)
if err != nil {
return delivery.Have{}, false, err
return nodeengine.HaveByNodeModuleValue{}, false, err
}
read, cancel := context.WithTimeout(ctx, q.timeout())
defer cancel()
bucket, err := hb.HaveBind(read)
if err != nil {
return delivery.Have{}, false, err
return nodeengine.HaveByNodeModuleValue{}, false, err
}
value, _, err := bucket.Get(read, key)
if errors.Is(err, ErrHaveAbsent) {
return delivery.Have{}, false, nil
return nodeengine.HaveByNodeModuleValue{}, false, nil
}
if err != nil {
return delivery.Have{}, false, err
return nodeengine.HaveByNodeModuleValue{}, false, err
}
h, err := delivery.DecodeHave(value)
h, err := nodeengineclient.DecodeHave(value)
if err != nil {
return delivery.Have{}, false, nil // a have no engine can read names nothing it runs
return nodeengine.HaveByNodeModuleValue{}, false, nil // a have no engine can read names nothing it runs
}
return h, true, nil
}
// nothingJudged is the health of a module a look judged no resource of: ok when it was looked at and the module applied
// and runs (nothing of it could be down), unknown otherwise.
func nothingJudged(f ModuleFact, looked bool) delivery.Health {
if looked && !f.Failed && (f.Phase == "" || f.Phase == string(delivery.PhaseRunning)) {
return delivery.HealthOK
func nothingJudged(f ModuleFact, looked bool) string {
if looked && !f.Failed && (f.Phase == "" || f.Phase == string(nodeengineclient.PhaseRunning)) {
return nodeengineclient.HealthOK
}
return delivery.HealthUnknown
}
// interimAnswers is the node's candidate answers in the form mesh-sdk's HaveNode still holds them, until the node-engine's
// have is written through the seat's contract types (the Have group of the go/delivery rewrite): the two shapes share
// their wire names, so the JSON carries one into the other.
func interimAnswers(in map[string]nodeengine.CandidateAnswer) map[string]delivery.CandidateAnswer {
if in == nil {
return nil
}
raw, err := json.Marshal(in)
if err != nil {
return nil
}
var out map[string]delivery.CandidateAnswer
if json.Unmarshal(raw, &out) != nil {
return nil
}
return out
return nodeengineclient.HealthUnknown
}
internal/link/have_nats_integration_test.go
+10 -10
View File
@@ -4,12 +4,12 @@ package link
import (
"context"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"github.com/nats-io/nats.go"
"github.com/novox/mesh-host/internal/buckets"
"sync/atomic"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
)
// 1132/AC8: against a real bus with the bucket node-engine_have, a node-engine that applies a declaration of two
@@ -21,9 +21,9 @@ import (
func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) {
conn, js := aBus(t)
const node = "haveok"
kv, err := js.KeyValue(delivery.HaveBucket)
kv, err := js.KeyValue(buckets.Have)
if err != nil {
if kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: delivery.HaveBucket, History: delivery.HaveHistory}); err != nil {
if kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buckets.Have, History: buckets.HaveHistory}); err != nil {
t.Fatal(err)
}
}
@@ -66,17 +66,17 @@ func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) {
first := revisions()
a, _ := kv.Get(node + ".a")
haveA, err := delivery.DecodeHave(a.Value())
if err != nil || haveA.Phase != delivery.PhaseRunning || haveA.Build != "b-1" || haveA.Health != delivery.HealthOK {
haveA, err := nodeengineclient.DecodeHave(a.Value())
if err != nil || haveA.Phase != nodeengineclient.PhaseRunning || haveA.Build != "b-1" || haveA.Health != nodeengineclient.HealthOK {
t.Fatalf("a holds %+v (%v)", haveA, err)
}
b, _ := kv.Get(node + ".b")
haveB, err := delivery.DecodeHave(b.Value())
if err != nil || haveB.Phase != delivery.PhaseFailed || haveB.Detail == "" {
haveB, err := nodeengineclient.DecodeHave(b.Value())
if err != nil || haveB.Phase != nodeengineclient.PhaseFailed || haveB.Detail == "" {
t.Fatalf("b holds %+v (%v)", haveB, err)
}
n, _ := kv.Get(node + ".node")
if hn, err := delivery.DecodeHaveNode(n.Value()); err != nil || len(hn.Windows) == 0 {
if hn, err := nodeengineclient.DecodeHaveNode(n.Value()); err != nil || len(hn.Windows) == 0 {
t.Fatalf("the node key holds %+v (%v)", hn, err)
}
internal/link/have_test.go
+77 -72
View File
@@ -5,6 +5,7 @@ import (
"encoding/json"
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
"github.com/novox/mesh-host/internal/compositions"
@@ -15,7 +16,6 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
)
@@ -104,11 +104,11 @@ func (l *haveLink) Set(_ context.Context, key string, value []byte, revision uin
return nil
}
func (l *haveLink) have(t *testing.T, key string) delivery.Have {
func (l *haveLink) have(t *testing.T, key string) nodeengine.HaveByNodeModuleValue {
t.Helper()
l.hmu.Lock()
defer l.hmu.Unlock()
h, err := delivery.DecodeHave(l.kv[key].value)
h, err := nodeengineclient.DecodeHave(l.kv[key].value)
if err != nil {
t.Fatalf("%s holds no have: %v", key, err)
}
@@ -121,9 +121,9 @@ func (l *haveLink) setCount() int {
return len(l.sets)
}
func (l *haveLink) put(t *testing.T, key string, h delivery.Have) {
func (l *haveLink) put(t *testing.T, key string, h nodeengine.HaveByNodeModuleValue) {
t.Helper()
raw, err := delivery.EncodeHave(h)
raw, err := nodeengineclient.EncodeHave(h)
if err != nil {
t.Fatal(err)
}
@@ -165,11 +165,11 @@ func TestAnAppliedModuleIsWrittenRunningAndAFailedOneFailed(t *testing.T) {
q, _ := aHaveQueue(l)
q.SayFacts(context.Background(), appliedModules())
a, b := l.have(t, "n1.a"), l.have(t, "n1.b")
if a.Phase != delivery.PhaseRunning || a.Build != "b-1" || a.Health != delivery.HealthOK {
if a.Phase != nodeengineclient.PhaseRunning || a.Build != "b-1" || a.Health != nodeengineclient.HealthOK {
t.Errorf("a: %+v", a)
}
if b.Phase != delivery.PhaseFailed || b.Build != "b-2" || !strings.Contains(b.Detail, "no space") ||
b.Health != delivery.HealthFailed {
if b.Phase != nodeengineclient.PhaseFailed || b.Build != "b-2" || !strings.Contains(b.Detail, "no space") ||
b.Health != nodeengineclient.HealthFailed {
t.Errorf("b: %+v", b)
}
if l.kv["n1.a"].rev != 1 {
@@ -180,14 +180,14 @@ func TestAnAppliedModuleIsWrittenRunningAndAFailedOneFailed(t *testing.T) {
// 1132/AC5: a have equal to the bucket's value is not written; a different one is written over the last revision.
func TestAHaveIsWrittenOnlyWhenItDiffers(t *testing.T) {
l := newHaveLink()
l.put(t, "n1.a", delivery.Have{V: sdkkeys.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.b", delivery.Have{V: sdkkeys.Version, Build: "b-0", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.a", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "b-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
l.put(t, "n1.b", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "b-0", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
q, _ := aHaveQueue(l)
q.SayFacts(context.Background(), appliedModules())
if l.kv["n1.a"].rev != 1 {
t.Errorf("a equal to the bucket was written again, to revision %d", l.kv["n1.a"].rev)
}
if l.kv["n1.b"].rev != 2 || l.have(t, "n1.b").Phase != delivery.PhaseFailed {
if l.kv["n1.b"].rev != 2 || l.have(t, "n1.b").Phase != nodeengineclient.PhaseFailed {
t.Errorf("b was not updated over its last revision: %+v at %d", l.have(t, "n1.b"), l.kv["n1.b"].rev)
}
before := l.setCount()
@@ -205,7 +205,7 @@ func TestAModulesHealthFollowsTheLook(t *testing.T) {
{Module: "c", Build: "b"}, {Module: "d", Build: "b"}, {Module: "e", Build: "b"}}}
q.SayFacts(context.Background(), rep)
for _, m := range []string{"a", "b", "c", "d", "e"} {
if h := l.have(t, "n1."+m).Health; h != delivery.HealthUnknown {
if h := l.have(t, "n1."+m).Health; h != nodeengineclient.HealthUnknown {
t.Errorf("%s with no look is %s, want unknown", m, h)
}
}
@@ -220,8 +220,8 @@ func TestAModulesHealthFollowsTheLook(t *testing.T) {
if !q.SayHealth(context.Background(), *h) {
t.Fatal("not said")
}
for m, want := range map[string]delivery.Health{"a": delivery.HealthWaits, "b": delivery.HealthFailed,
"c": delivery.HealthWaits, "d": delivery.HealthUnknown, "e": delivery.HealthOK} {
for m, want := range map[string]string{"a": nodeengineclient.HealthWaits, "b": nodeengineclient.HealthFailed,
"c": nodeengineclient.HealthWaits, "d": nodeengineclient.HealthUnknown, "e": nodeengineclient.HealthOK} {
if got := l.have(t, "n1."+m).Health; got != want {
t.Errorf("%s is %s, want %s", m, got, want)
}
@@ -239,7 +239,7 @@ func TestAModulesHealthFollowsTheLook(t *testing.T) {
}
h2.Resources = []ResourceHealth{{Module: "a", Resource: "a.server", State: StateHealthy}}
q.SayHealth(context.Background(), h2)
if got := l.have(t, "n1.a"); got.Health != delivery.HealthOK || got.Detail != "" {
if got := l.have(t, "n1.a"); got.Health != nodeengineclient.HealthOK || got.Detail != "" {
t.Errorf("a after a healthy look is %s, saying %q", got.Health, got.Detail)
}
}
@@ -255,14 +255,19 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
rep.Filters = []Filter{{Where: "inet mesh", Owner: "mesh"}}
rep.Profile = map[string]any{"cores": 8}
q.SayFacts(context.Background(), rep)
got, err := delivery.DecodeHaveNode(l.kv["n1.node"].value)
got, err := nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value)
if err != nil {
t.Fatal(err)
}
enc := func(v any) json.RawMessage { b, _ := json.Marshal(v); return b }
want := delivery.HaveNode{V: sdkkeys.Version, Windows: enc(rep.Windows), Busy: enc(rep.Busy), Strays: enc(rep.Strays),
Filters: enc(rep.Filters), Profile: enc(rep.Profile)}
if !delivery.SameHaveNode(got, want) {
enc := func(v any) []map[string]any {
var out []map[string]any
b, _ := json.Marshal(v)
_ = json.Unmarshal(b, &out)
return out
}
want := nodeengine.HaveByNodeNodeValue{V: sdkkeys.Version, Windows: enc(rep.Windows), Busy: enc(rep.Busy), Strays: enc(rep.Strays),
Filters: enc(rep.Filters), Profile: rep.Profile}
if !nodeengineclient.SameHaveNode(got, want) {
t.Errorf("the node key holds %+v, want %+v", got, want)
}
@@ -279,14 +284,14 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
rep.Outward = []string{"eth0"}
rep.Health = &Health{Contract: NodeKeyContract}
q.SayFacts(context.Background(), rep)
if got, err = delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || got.Record == nil {
if got, err = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); err != nil || got.Record == nil {
t.Fatalf("the node key holds no record: %+v, %v", got, err)
}
wantRecord := delivery.NodeRecord{V: sdkkeys.Version, Node: "n1", TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk",
wantRecord := nodeengine.NodeRecord{V: sdkkeys.Version, Node: "n1", TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk",
Certificate: "cert", Trusts: keys.Trusts, Capabilities: []string{"apt", "bus-state-calls", "ledger-calls", "systemd"}, OutwardLinks: []string{"eth0"}, NetworksSaid: true,
Networks: []catalogue.AttachedNetwork{{Link: "eth0", Network: "192.0.2.0/24"}}, Contract: NodeKeyContract,
Networks: []nodeengine.AttachedNetwork{{Link: "eth0", Network: "192.0.2.0/24"}}, Contract: NodeKeyContract,
ManifestReads: catalogue.ManifestReads()} // what it reads of a manifest: its contract's (ADR 0614 §4)
if !delivery.SameNodeRecord(*got.Record, wantRecord) {
if !nodeengineclient.SameNodeRecord(*got.Record, wantRecord) {
t.Errorf("record = %+v, want %+v", *got.Record, wantRecord)
}
// Among them, that it fills a seat's reach from the seat holder's listen (ADR 0632 §1).
@@ -307,7 +312,7 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
}
l.statedErr, l.stated = nil, map[string]string{}
q.SayFacts(context.Background(), rep)
if got, _ = delivery.DecodeHaveNode(l.kv["n1.node"].value); l.kv["n1.node"].rev != at+1 || !slices.Equal(got.Record.Capabilities, []string{"apt", "systemd"}) {
if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); l.kv["n1.node"].rev != at+1 || !slices.Equal(got.Record.Capabilities, []string{"apt", "systemd"}) {
t.Errorf("a runner stating no calls: revision %d (was %d), capabilities %v", l.kv["n1.node"].rev, at, got.Record.Capabilities)
}
at++
@@ -316,13 +321,13 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
at++
keys.Certificate = "cert-2"
q.SayFacts(context.Background(), rep)
if got, _ = delivery.DecodeHaveNode(l.kv["n1.node"].value); l.kv["n1.node"].rev != at+1 || got.Record.Certificate != "cert-2" {
if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); l.kv["n1.node"].rev != at+1 || got.Record.Certificate != "cert-2" {
t.Errorf("a new certificate: revision %d (was %d), record %+v", l.kv["n1.node"].rev, at, got.Record)
}
bare := rep
bare.Health = nil
q.SayFacts(context.Background(), bare)
if got, _ = delivery.DecodeHaveNode(l.kv["n1.node"].value); got.Record.Contract != NodeKeyContract {
if got, _ = nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); got.Record.Contract != NodeKeyContract {
t.Errorf("a report without a health statement dropped the contract: %+v", got.Record)
}
}
@@ -342,7 +347,7 @@ func TestAFailedHaveWriteIsSaidAndWrittenAgain(t *testing.T) {
l.failing = false
l.hmu.Unlock()
q.SayFacts(context.Background(), appliedModules())
if l.have(t, "n1.a").Phase != delivery.PhaseRunning {
if l.have(t, "n1.a").Phase != nodeengineclient.PhaseRunning {
t.Error("the next apply did not write again")
}
}
@@ -407,7 +412,7 @@ func TestAFailedBindOrWriteEndsThePassAndReturnsInTime(t *testing.T) {
if l.setCount() == 0 {
t.Fatal("a bind that met no responders was not written again")
}
if l.have(t, "n1.a").Phase != delivery.PhaseRunning {
if l.have(t, "n1.a").Phase != nodeengineclient.PhaseRunning {
t.Error("the have written again does not say running")
}
}
@@ -426,13 +431,13 @@ func TestAModuleLeavingTheDeclarationIsWrittenRemoved(t *testing.T) {
revA, revB := l.kv["n1.a"].rev, l.kv["n1.b"].rev
q.SayFacts(context.Background(), onlyA())
b := l.have(t, "n1.b")
if b.Phase != delivery.PhaseRemoved || b.Build != "b-2" {
if b.Phase != nodeengineclient.PhaseRemoved || b.Build != "b-2" {
t.Errorf("b: %+v", b)
}
if l.kv["n1.b"].rev != revB+1 {
t.Errorf("b was written at revision %d, want one over %d", l.kv["n1.b"].rev, revB)
}
if l.kv["n1.a"].rev != revA || l.have(t, "n1.a").Phase != delivery.PhaseRunning {
if l.kv["n1.a"].rev != revA || l.have(t, "n1.a").Phase != nodeengineclient.PhaseRunning {
t.Errorf("a changed: %+v at %d", l.have(t, "n1.a"), l.kv["n1.a"].rev)
}
before := l.setCount()
@@ -449,11 +454,11 @@ func TestARemovedModuleAddedBackIsWrittenWithItsNewPhase(t *testing.T) {
q.SayFacts(context.Background(), appliedModules())
q.SayFacts(context.Background(), onlyA())
q.SayFacts(context.Background(), appliedModules())
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseFailed || b.Build != "b-2" {
if b := l.have(t, "n1.b"); b.Phase != nodeengineclient.PhaseFailed || b.Build != "b-2" {
t.Errorf("b after it came back: %+v", b)
}
q.SayFacts(context.Background(), onlyA())
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
if b := l.have(t, "n1.b"); b.Phase != nodeengineclient.PhaseRemoved {
t.Errorf("b removed again: %+v", b)
}
}
@@ -462,11 +467,11 @@ func TestARemovedModuleAddedBackIsWrittenWithItsNewPhase(t *testing.T) {
// bucket is not created again.
func TestAModuleNeverWrittenIsNotWrittenRemoved(t *testing.T) {
l := newHaveLink()
l.put(t, "n1.c", delivery.Have{V: sdkkeys.Version, Build: "b-9", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.c", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "b-9", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
q, _ := aHaveQueue(l)
q.SayFacts(context.Background(), appliedModules())
q.SayFacts(context.Background(), onlyA())
if c := l.have(t, "n1.c"); c.Phase != delivery.PhaseRunning || l.kv["n1.c"].rev != 1 {
if c := l.have(t, "n1.c"); c.Phase != nodeengineclient.PhaseRunning || l.kv["n1.c"].rev != 1 {
t.Errorf("c, which this node-engine never wrote, changed: %+v at %d", c, l.kv["n1.c"].rev)
}
// a key that vanished from the bucket is not created by the removal
@@ -495,7 +500,7 @@ func TestARefusedRemovalIsWrittenByTheNextPass(t *testing.T) {
l.failing = false
l.hmu.Unlock()
q.SayFacts(context.Background(), onlyA())
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
if b := l.have(t, "n1.b"); b.Phase != nodeengineclient.PhaseRemoved {
t.Errorf("b: %+v", b)
}
}
@@ -504,13 +509,13 @@ func TestARefusedRemovalIsWrittenByTheNextPass(t *testing.T) {
// before: one of them the new declaration lacks is written removed, one that is not in the bucket is not created.
func TestARestartedNodeEngineWritesTheRemovalFromTheReportsBefore(t *testing.T) {
l := newHaveLink()
l.put(t, "n1.a", delivery.Have{V: sdkkeys.Version, Build: "b-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.b", delivery.Have{V: sdkkeys.Version, Build: "b-2", Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
l.put(t, "n1.a", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "b-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
l.put(t, "n1.b", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "b-2", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK})
q, _ := aHaveQueue(l) // fresh: nothing written by this process
r := onlyA()
r.Before = []string{"a", "b", "c"}
q.SayFacts(context.Background(), r)
if b := l.have(t, "n1.b"); b.Phase != delivery.PhaseRemoved {
if b := l.have(t, "n1.b"); b.Phase != nodeengineclient.PhaseRemoved {
t.Errorf("b: %+v", b)
}
if _, there := l.kv["n1.c"]; there {
@@ -563,24 +568,24 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
dig := "sha256:" + strings.Repeat("b", 64)
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig})
h := l.have(t, "n1.web")
if h.FP != fp || h.Composed != dig || h.Phase != delivery.PhaseRunning || h.Health != delivery.HealthUnknown {
if h.Fp != fp || h.Composed != dig || h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("the have after the apply: %+v", h)
}
// Starting: it waits. Past its start window the check fails: failed.
q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateStarting}))
if got := l.have(t, "n1.web"); got.Health != delivery.HealthWaits || got.FP != fp || got.Composed != dig {
if got := l.have(t, "n1.web"); got.Health != nodeengineclient.HealthWaits || got.Fp != fp || got.Composed != dig {
t.Fatalf("while it starts: %+v", got)
}
q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateUnhealthy}))
if got := l.have(t, "n1.web"); got.Health != delivery.HealthFailed || got.FP != fp || got.Composed != dig {
if got := l.have(t, "n1.web"); got.Health != nodeengineclient.HealthFailed || got.Fp != fp || got.Composed != dig {
t.Fatalf("a failing check after the start window: %+v", got)
}
// The bus's user list written alone rides on the module's have, and an apply after it keeps it (ADR 0547 §8).
w := delivery.UserListWritten{Source: "n1.web." + fp, Revision: 4, Wanted: dig, Digest: dig, Users: 3,
w := nodeengine.UserListWritten{Source: "n1.web." + fp, Revision: 4, Wanted: dig, Digest: dig, Users: 3,
At: time.Date(2026, 10, 25, 21, 0, 0, 0, time.UTC)}
q.NoteUserList(context.Background(), "web", w)
if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Revision != 4 || got.FP != fp {
if got := l.have(t, "n1.web"); got.UserList == nil || got.UserList.Revision != 4 || got.Fp != fp {
t.Fatalf("the user list is not in the have: %+v", got)
}
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig})
@@ -589,24 +594,24 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
}
// A failed apply waiting on a store's holder carries the pull it waits on in its have (ENG-9, read by node_haves).
wait := &delivery.PullWait{Image: "reg/x@sha256:1", Seat: "container-image-registry", Holder: "distribution",
wait := &nodeengine.PullWait{Image: "reg/x@sha256:1", Seat: "container-image-registry", Holder: "distribution",
Began: time.Date(2026, 11, 4, 14, 0, 0, 0, time.UTC), Retried: "distribution settled"}
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Failed: true, Detail: "the pull failed", PullWait: wait})
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseFailed || got.PullWait == nil || got.PullWait.Holder != "distribution" {
if got := l.have(t, "n1.web"); got.Phase != nodeengineclient.PhaseFailed || got.PullWait == nil || got.PullWait.Holder != "distribution" {
t.Fatalf("the failed have does not say the pull it waits on: %+v", got)
}
// A module held for a failed backup says so and what it keeps running; removed is written once, over its key.
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Phase: string(delivery.PhaseHeld), Detail: "the repository is locked"})
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseHeld || got.Detail == "" {
q.NoteModule(context.Background(), ModuleFact{Module: "web", FP: fp, Composed: dig, Phase: string(nodeengineclient.PhaseHeld), Detail: "the repository is locked"})
if got := l.have(t, "n1.web"); got.Phase != nodeengineclient.PhaseHeld || got.Detail == "" {
t.Fatalf("held: %+v", got)
}
q.NoteRemoved(context.Background(), "web", "")
if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved {
if got := l.have(t, "n1.web"); got.Phase != nodeengineclient.PhaseRemoved {
t.Fatalf("removed: %+v", got)
}
q.NoteNode(context.Background(), dig, compositions.Seals{"core-box#sealed": "core"})
if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig || len(n.SealedAt) != 1 || n.SealedAt["core-box#sealed"] != "core" {
if n, err := nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig || len(n.SealedAt) != 1 || n.SealedAt["core-box#sealed"] != "core" {
t.Fatalf("the node key, with its own composition's sealed values by slot: %+v %v", n, err)
}
if n, ok := q.ReadNodeHave(context.Background()); !ok || n.Composed != dig || len(n.SealedAt) != 1 {
@@ -615,11 +620,11 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth(
// The node's answers to placement candidates ride in the same key (ADR 0559 §4), and go when no group is answered.
answers := map[string]nodeengine.CandidateAnswer{"p-1": {Proposal: "p-1", Revision: 1, Verdict: nodeengine.CandidateAnswerVerdictSame, Same: []string{"web"}}}
q.NoteCandidates(context.Background(), answers)
if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["p-1"].Proposal != "p-1" || n.Composed != dig {
if n, err := nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["p-1"].Proposal != "p-1" || n.Composed != dig {
t.Fatalf("the node key with its answers: %+v %v", n, err)
}
q.NoteCandidates(context.Background(), nil)
if n, _ := delivery.DecodeHaveNode(l.kv["n1.node"].value); n.Candidates != nil {
if n, _ := nodeengineclient.DecodeHaveNode(l.kv["n1.node"].value); n.Candidates != nil {
t.Fatalf("an answer left after its group went: %+v", n.Candidates)
}
}
@@ -633,16 +638,16 @@ func TestAModuleWithNothingToJudgeIsOkOnceLookedAt(t *testing.T) {
q, _ := aHaveQueue(l)
q.SayFacts(context.Background(), Report{Modules: []ModuleFact{{Module: "mesh-host", Build: "b"}, {Module: "web", Build: "b"},
{Module: "broken", Build: "b", Failed: true, Detail: "x"}}})
if h := l.have(t, "n1.mesh-host").Health; h != delivery.HealthUnknown {
if h := l.have(t, "n1.mesh-host").Health; h != nodeengineclient.HealthUnknown {
t.Fatalf("before any look: %s", h)
}
if !q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) {
t.Fatal("not said")
}
if h := l.have(t, "n1.mesh-host").Health; h != delivery.HealthOK {
if h := l.have(t, "n1.mesh-host").Health; h != nodeengineclient.HealthOK {
t.Fatalf("a running module with nothing to judge after a look: %s", h)
}
if h := l.have(t, "n1.broken").Health; h == delivery.HealthOK {
if h := l.have(t, "n1.broken").Health; h == nodeengineclient.HealthOK {
t.Fatalf("a failed module was said ok")
}
}
@@ -654,17 +659,17 @@ func TestAModuleSaidAgainOverAnOlderEnginesUnknownIsWrittenWithThisEnginesHealth
l := newHaveLink()
fp := "src2:" + strings.Repeat("a", 64)
dig := "sha256:" + strings.Repeat("b", 64)
l.put(t, "n1.xorg", delivery.Have{V: sdkkeys.Version, FP: fp, Composed: dig, Phase: delivery.PhaseRunning,
Health: delivery.HealthUnknown})
l.put(t, "n1.xorg", nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Fp: fp, Composed: dig, Phase: nodeengineclient.PhaseRunning,
Health: nodeengineclient.HealthUnknown})
q, _ := aHaveQueue(l)
if !q.SayHealth(context.Background(), *healthOf(map[string]string{"web": StateHealthy})) {
t.Fatal("not said")
}
if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.xorg"); h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("a module the session does not hold was written: %+v", h)
}
q.NoteModule(context.Background(), ModuleFact{Module: "xorg", FP: fp, Composed: dig})
if h := l.have(t, "n1.xorg"); h.Health != delivery.HealthOK || h.FP != fp || h.Composed != dig || h.Phase != delivery.PhaseRunning {
if h := l.have(t, "n1.xorg"); h.Health != nodeengineclient.HealthOK || h.Fp != fp || h.Composed != dig || h.Phase != nodeengineclient.PhaseRunning {
t.Fatalf("said again after a look: %+v", h)
}
}
@@ -679,26 +684,26 @@ func TestAHealthIsNotJudgedWhileAModuleIsApplied(t *testing.T) {
old := ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("a", 64), Composed: "sha256:" + strings.Repeat("1", 64)}
q.NoteModule(ctx, old)
q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy}))
if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthFailed {
if h := l.have(t, "n1.ctl"); h.Health != nodeengineclient.HealthFailed {
t.Fatalf("the running module's failed look was not said: %+v", h)
}
next := ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("b", 64), Composed: "sha256:" + strings.Repeat("2", 64)}
applying := next
applying.Phase = string(delivery.PhaseApplying)
applying.Phase = string(nodeengineclient.PhaseApplying)
q.NoteModule(ctx, applying)
if h := l.have(t, "n1.ctl"); h.Phase != delivery.PhaseApplying || h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.ctl"); h.Phase != nodeengineclient.PhaseApplying || h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("applying carried the last look's health: %+v", h)
}
q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy}))
if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.ctl"); h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("a look during the apply judged it: %+v", h)
}
q.NoteModule(ctx, next)
if h := l.have(t, "n1.ctl"); h.Phase != delivery.PhaseRunning || h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.ctl"); h.Phase != nodeengineclient.PhaseRunning || h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("running before a look after the apply: %+v", h)
}
q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateHealthy}))
if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthOK {
if h := l.have(t, "n1.ctl"); h.Health != nodeengineclient.HealthOK {
t.Fatalf("the look after the apply was not taken: %+v", h)
}
}
@@ -711,7 +716,7 @@ func TestAHealthIsNotCarriedOverToAnotherComposition(t *testing.T) {
q.NoteModule(ctx, ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("a", 64)})
q.SayHealth(ctx, *healthOf(map[string]string{"ctl": StateUnhealthy}))
q.NoteModule(ctx, ModuleFact{Module: "ctl", FP: "src2:" + strings.Repeat("b", 64)})
if h := l.have(t, "n1.ctl"); h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.ctl"); h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("the old composition's look judged the new one: %+v", h)
}
}
@@ -724,11 +729,11 @@ func TestADeclarationReportMovesAHealthAndARemovalForgetsIt(t *testing.T) {
ctx := context.Background()
q.SayFacts(ctx, Report{Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1"}},
Health: healthOf(map[string]string{"a": StateUnhealthy})})
if h := l.have(t, "n1.a"); h.Health != delivery.HealthFailed {
if h := l.have(t, "n1.a"); h.Health != nodeengineclient.HealthFailed {
t.Fatalf("the failed look was not said: %+v", h)
}
q.SayFacts(ctx, Report{Applied: []string{"a.x"}, Modules: []ModuleFact{{Module: "a", Build: "b-1", FP: "src2:" + strings.Repeat("c", 64)}}})
if h := l.have(t, "n1.a"); h.Health != delivery.HealthUnknown {
if h := l.have(t, "n1.a"); h.Health != nodeengineclient.HealthUnknown {
t.Fatalf("the old composition's look judged the new one: %+v", h)
}
select {
@@ -750,18 +755,18 @@ func TestADeclarationReportMovesAHealthAndARemovalForgetsIt(t *testing.T) {
// key that does not exist is not made up and is said once; a writer that drops the list again is written over.
func TestAUserListReachesTheHaveOfAModuleNoApplyHereSays(t *testing.T) {
l := newHaveLink()
older := delivery.Have{V: sdkkeys.Version, Build: "build-1", Phase: delivery.PhaseRunning, Health: delivery.HealthOK}
older := nodeengine.HaveByNodeModuleValue{V: sdkkeys.Version, Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}
l.put(t, "n1.nats", older)
q, said := aHaveQueue(l)
dig := "sha256:" + strings.Repeat("c", 64)
w := delivery.UserListWritten{Source: "n1.nats", Revision: 7, Wanted: dig, Digest: dig, Users: 58,
w := nodeengine.UserListWritten{Source: "n1.nats", Revision: 7, Wanted: dig, Digest: dig, Users: 58,
At: time.Date(2026, 10, 26, 12, 26, 51, 0, time.UTC)}
q.NoteUserList(context.Background(), "nats", w)
got := l.have(t, "n1.nats")
if got.UserList == nil || got.UserList.Users != 58 || got.UserList.Digest != dig {
t.Fatalf("the user list is not in the have: %+v", got)
}
if got.Build != "build-1" || got.Phase != delivery.PhaseRunning || got.Health != delivery.HealthOK {
if got.Build != "build-1" || got.Phase != nodeengineclient.PhaseRunning || got.Health != nodeengineclient.HealthOK {
t.Fatalf("the merge lost the key's own fields: %+v", got)
}
rev := l.kv["n1.nats"].rev
@@ -794,7 +799,7 @@ func TestAUserListReachesTheHaveOfAModuleNoApplyHereSays(t *testing.T) {
}
fp := "src2:" + strings.Repeat("a", 64)
q.NoteModule(context.Background(), ModuleFact{Module: "other", FP: fp, Composed: dig})
if got := l.have(t, "n1.other"); got.UserList == nil || got.FP != fp {
if got := l.have(t, "n1.other"); got.UserList == nil || got.Fp != fp {
t.Fatalf("the first have of the module lost the list: %+v", got)
}
}
internal/link/health_test.go
+4 -5
View File
@@ -3,12 +3,11 @@ package link
import (
"context"
"encoding/json"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"slices"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A health statement is said on the link open now, as the machine's own word on its own subject, and
@@ -60,7 +59,7 @@ func TestAHealthStatementIsSaidOnTheLinkOpenNow(t *testing.T) {
since := time.Date(2026, 10, 29, 20, 0, 0, 0, time.UTC)
look := func(streak int, writer string) Health {
return Health{Contract: 7, At: time.Now(), Network: &NetworkHealth{State: StateUnhealthy, Since: since, Parts: []NetworkPart{
{Part: "resolv-conf", State: StateUnhealthy, Writer: writer, Owner: "dnsmasq", Since: since, Streak: streak}}},
{Part: "resolv-conf", State: StateUnhealthy, Writer: writer, Owner: "dnsmasq", Since: since, Streak: int64(streak)}}},
Units: &UnitsHealth{State: "degraded", Failed: []FailedUnit{{Unit: "cups.service", Scope: "system", Since: since}}},
Resources: []ResourceHealth{
{Module: "brother", Resource: "brother.scanner", Kind: KindAccount, Account: "jochen", State: StateUnhealthy,
@@ -121,12 +120,12 @@ func TestAHealthStatementIsSaidOnTheLinkOpenNow(t *testing.T) {
t.Fatalf("step %d: wrote %v (want %v), %v", i, wrote, step.wrote, err)
}
}
got, err := delivery.DecodeNodeHealth(store.values["anchor"])
got, err := nodeengineclient.DecodeHealth(store.values["anchor"])
if err != nil || got.Node != "anchor" || got.Network.Parts[0].Writer != "openvpn" || !got.Peers[0].Seen || got.At.IsZero() {
t.Fatalf("the key holds %+v, %v", got, err)
}
// The key's bytes are mesh-sdk's layout: decoded and encoded again by it, the same bytes.
if again, err := delivery.EncodeNodeHealth(got); err != nil || string(again) != string(store.values["anchor"]) {
if again, err := nodeengineclient.EncodeHealth(got); err != nil || string(again) != string(store.values["anchor"]) {
t.Fatalf("the key's bytes are not the sdk's layout: %s\n%s (%v)", store.values["anchor"], again, err)
}
if store.puts != 3 {
internal/link/messages.go
+11 -12
View File
@@ -1,11 +1,10 @@
package link
import (
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/health"
"github.com/novox/mesh-host/internal/compositions"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
// The wire formats shared with the control plane, which defines them separately because this
@@ -304,10 +303,10 @@ const ReasonUsedAsFound = "used as found:"
// The machine's networking and its service managers in a statement are the node's health report's parts (mesh-sdk
// delivery, issue 774): one layout for the message, the report and the watchman that reads it.
type (
NetworkHealth = delivery.NetworkHealth
NetworkPart = delivery.NetworkPart
UnitsHealth = delivery.UnitsHealth
FailedUnit = delivery.FailedUnit
NetworkHealth = nodeengine.NetworkHealth
NetworkPart = nodeengine.NetworkPart
UnitsHealth = nodeengine.UnitsHealth
FailedUnit = nodeengine.FailedUnit
)
// The states a long-running resource is said in (ADR 0240 §4).
@@ -640,12 +639,12 @@ type ModuleFact struct {
// Judgement and Waits are, in phase waiting-yes, the judgement of the move and the composition waiting (ADR 0547 §3).
Judgement string
Waits string
// BuildAlone says, in phase waiting-yes, the move changes the build alone (ADR 0601 §1, delivery.Have.BuildAlone).
// BuildAlone says, in phase waiting-yes, the move changes the build alone (ADR 0601 §1, nodeengine.HaveByNodeModuleValue.BuildAlone).
BuildAlone bool
// Kept is what the node keeps of the module as it runs, for a manifest that does not say it yet (delivery.Have.Kept).
// Kept is what the node keeps of the module as it runs, for a manifest that does not say it yet (nodeengine.HaveByNodeModuleValue.Kept).
Kept []string
// Sealed is the values the composition it names carries sealed to this node's key, by slot (delivery.Have.SealedAt).
// Sealed is the values the composition it names carries sealed to this node's key, by slot (nodeengine.HaveByNodeModuleValue.SealedAt).
Sealed compositions.Seals
// PullWait is, on a failed fact, the pull from a store its apply waits on (delivery.Have.PullWait, issue 1966 ENG-9).
PullWait *delivery.PullWait
// PullWait is, on a failed fact, the pull from a store its apply waits on (nodeengine.HaveByNodeModuleValue.PullWait, issue 1966 ENG-9).
PullWait *nodeengine.PullWait
}
internal/link/nodehealth.go
+16 -13
View File
@@ -4,13 +4,16 @@ import (
"context"
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/keys"
"github.com/novox/mesh-host/internal/buckets"
"slices"
"sort"
"strings"
"time"
"git.novox.be/novox/mesh-sdk/go/declaration"
"git.novox.be/novox/mesh-sdk/go/delivery"
"github.com/nats-io/nats.go"
)
@@ -30,17 +33,17 @@ import (
// - node-account-fallback: Fallback, the engine's fall back to its old login during a rotation (ADR 0583 §6).
//
// The report's layout is mesh-sdk's (delivery.NodeHealth): the bucket, its version and its parts.
// The report's layout is mesh-sdk's (nodeengine.HealthByNodeValue): the bucket, its version and its parts.
const (
HealthBucket = delivery.NodeHealthBucket
NodeHealthVersion = delivery.NodeHealthVersion
HealthBucket = buckets.Health
NodeHealthVersion = keys.Version
)
type (
NodeHealth = delivery.NodeHealth
PeerSeen = delivery.PeerSeen
NodeAccount = delivery.NodeAccount
NodeDirectory = delivery.NodeDirectory
NodeHealth = nodeengine.HealthByNodeValue
PeerSeen = nodeengine.PeerSeen
NodeAccount = nodeengine.NodeAccount
NodeDirectory = nodeengine.NodeDirectory
)
// ReasonReloginNeeded starts an account's reason when its session needs a new login (accounts.ReasonRelogin).
@@ -83,7 +86,7 @@ const TowardHub = "hub"
// streaks, each unhealthy part naming the machines it points at (machines), its machine units, and each account and
// directory of its resources that the six alerts read.
func NodeHealthOf(node string, h Health, peers []PeerSeen, machines Machines) NodeHealth {
out := NodeHealth{V: NodeHealthVersion, Node: node, Units: h.Units, Contract: h.Contract}
out := NodeHealth{V: NodeHealthVersion, Node: node, Units: h.Units, Contract: int64(h.Contract)}
if len(peers) > 0 {
out.Peers = append([]PeerSeen(nil), peers...)
sort.Slice(out.Peers, func(i, k int) bool { return out.Peers[i].Key < out.Peers[k].Key })
@@ -108,7 +111,7 @@ func NodeHealthOf(node string, h Health, peers []PeerSeen, machines Machines) No
}
if r.Kind == KindAccount && r.Root == declaration.RootNever {
// The account declared never to become root without a person, judged (ADR 0266): what DA and D-root read.
out.Root = append(out.Root, delivery.RootVerdict{Account: r.Account, Module: r.Module, Resource: r.Resource,
out.Root = append(out.Root, nodeengine.RootAccountVerdict{Account: r.Account, Module: r.Module, Resource: r.Resource,
State: r.State, Reason: r.Reason, Since: r.Since})
}
if r.Kind == KindDirectory && strings.HasPrefix(r.Reason, ReasonUsedAsFound) {
@@ -191,16 +194,16 @@ func writeNodeHealth(ctx context.Context, store HealthStore, last **NodeHealth,
return false, fmt.Errorf("the health report of %s cannot be read: %w", h.Node, err)
}
if raw != nil {
if was, err := delivery.DecodeNodeHealth(raw); err == nil {
if was, err := nodeengineclient.DecodeHealth(raw); err == nil {
*last = &was
}
}
}
if *last != nil && delivery.SameNodeHealth(**last, h) {
if *last != nil && nodeengineclient.SameHealth(**last, h) {
return false, nil
}
h.At = now().UTC()
raw, err := delivery.EncodeNodeHealth(h)
raw, err := nodeengineclient.EncodeHealth(h)
if err != nil {
return false, err
}
internal/link/record.go
+8 -6
View File
@@ -2,6 +2,7 @@ package link
import (
"encoding/json"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
sdkkeys "git.novox.be/novox/mesh-sdk/go/keys"
"slices"
"sort"
@@ -9,11 +10,10 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// The node's record (novox/hq ADR 0555 §3, issue 1126): what the machine observes of itself, written in its own have key
// `<node>.node` (delivery.HaveNode.Record) with its facts, so every node and the parts that compose read it from the bus
// `<node>.node` (nodeengine.HaveByNodeNodeValue.Record) with its facts, so every node and the parts that compose read it from the bus
// and hear its change as the key's. Each value is the one the controller took from this node's enrolment and reports
// until now (its profile's present capabilities and networks, its outward links, its health statement's contract, and
// the public halves of the keys it enrolled with), read here from the same sources, never from the controller.
@@ -44,7 +44,7 @@ var readsStatedAt = time.Now().UTC()
// recordOf is the node's record from one facts report and its keys; nil for a report that carries no profile (one that
// says nothing of the machine, such as a rekey), so the record last written stands.
func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
func recordOf(node string, r Report, keys RecordKeys) *nodeengine.NodeRecord {
if r.Profile == nil {
return nil
}
@@ -61,7 +61,7 @@ func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
if err != nil || json.Unmarshal(raw, &p) != nil {
return nil
}
rec := &delivery.NodeRecord{V: sdkkeys.Version, Node: node, TunnelKey: keys.TunnelKey, SealingKey: keys.SealingKey,
rec := &nodeengine.NodeRecord{V: sdkkeys.Version, Node: node, TunnelKey: keys.TunnelKey, SealingKey: keys.SealingKey,
ServingKey: keys.ServingKey, Certificate: keys.Certificate, Authority: keys.Authority, Trusts: keys.Trusts,
ManifestReads: manifestReads()}
for _, c := range p.Capabilities {
@@ -77,13 +77,15 @@ func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
sort.Strings(rec.Capabilities)
if p.Machine != nil {
rec.NetworksSaid = true
rec.Networks = p.Machine.Networks
for _, n := range p.Machine.Networks {
rec.Networks = append(rec.Networks, nodeengine.AttachedNetwork{Link: n.Link, Network: n.Network})
}
}
if len(r.Outward) > 0 {
rec.OutwardLinks = append([]string(nil), r.Outward...)
}
if r.Health != nil {
rec.Contract = r.Health.Contract
rec.Contract = int64(r.Health.Contract)
}
return rec
}
internal/link/rotate.go
+5 -4
View File
@@ -3,6 +3,7 @@ package link
import (
"context"
"fmt"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"sync"
"time"
@@ -35,9 +36,9 @@ type Rotator struct {
failures int
// fellBack is the key's `since` a fallback was made for: the same rotation is not tried again.
fellBackSince string
// fallen is the fall back the node's health report carries (delivery.LoginFallback), from which the watchman raises
// fallen is the fall back the node's health report carries (nodeengine.LoginFallback), from which the watchman raises
// node-account-fallback (ADR 0583 §6), until the vault's key of this node says anything new.
fallen *delivery.LoginFallback
fallen *nodeengine.LoginFallback
switchedSince string
}
@@ -162,13 +163,13 @@ func (r *Rotator) AuthFailed(login string) {
if r.Now != nil {
now = r.Now
}
r.fallen = &delivery.LoginFallback{Node: r.Logins.node, From: login, Why: text, At: now().UTC()}
r.fallen = &nodeengine.LoginFallback{Node: r.Logins.node, From: login, Why: text, At: now().UTC()}
r.old, r.failures = "", 0
r.fellBackSince = r.switchedSince
}
// Fallback is the fall back the node's health report carries now, nil when there is none.
func (r *Rotator) Fallback() *delivery.LoginFallback {
func (r *Rotator) Fallback() *nodeengine.LoginFallback {
if r == nil {
return nil
}
internal/link/rotate_test.go
+3 -1
View File
@@ -3,6 +3,8 @@ package link
import (
"context"
"errors"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"slices"
"strings"
"testing"
@@ -140,7 +142,7 @@ func TestTheOldValueGoesAtRotationDoneAbandonSwitchesBackAndFailuresFallBack(t *
t.Fatalf("after the bound: %s, fall back %+v", cur, fb)
}
// The health report carries it, from which the watchman raises node-account-fallback.
if _, err := delivery.EncodeNodeHealth(delivery.NodeHealth{Node: "ace", Fallback: fb}); err != nil {
if _, err := nodeengineclient.EncodeHealth(nodeengine.HealthByNodeValue{Node: "ace", Fallback: fb}); err != nil {
t.Errorf("the fall back is no health report's: %v", err)
}
steps := len(f.steps)
internal/link/session.go
+1 -2
View File
@@ -12,7 +12,6 @@ import (
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/health"
)
@@ -23,7 +22,7 @@ import (
// receives, orders or reconciles a whole declaration.
type Session struct {
// Fallback is the engine's fall back to its old login its health report carries now (Rotator.Fallback); nil says none.
Fallback func() *delivery.LoginFallback
Fallback func() *nodeengine.LoginFallback
// linkedUser is the bus user the link open now authenticated as, guarded by mu (WhenLinkedAs).
linkedUser string
// Membership is the node and the mesh's key a controller's ask is verified with.