H-17: the node-engine's own name is read from its applied record or first composing
Discovery, the finding writer and the vault login names read it; no fixed name. Builds: design 47 G §527, §634 (A-149)
This commit is contained in:
@@ -1,12 +1,16 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/compositions"
|
||||
)
|
||||
|
||||
// appliedStore keeps every composition the agent applied, one file per module (the node's own as `node.json`), in the
|
||||
@@ -105,3 +109,19 @@ func (s appliedStore) Drop(module string) error {
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// ownModuleApplied is the module of the composition the store holds at the fingerprint fp, the engine's own build's
|
||||
// (design 47 G §527, A-149): "" when fp is none or no record holds it.
|
||||
func ownModuleApplied(s appliedStore, fp string) string {
|
||||
if fp == "" {
|
||||
return ""
|
||||
}
|
||||
all, _ := s.Load()
|
||||
for module, raw := range all {
|
||||
var c compositions.Composed
|
||||
if module != agent.NodeModule && json.Unmarshal(raw, &c) == nil && c.FP == fp && c.Module != "" {
|
||||
return c.Module
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
@@ -87,15 +87,16 @@ func followLogins(ctx context.Context, session *link.Session, node string, hear
|
||||
}
|
||||
}
|
||||
|
||||
// loginSecretModule is the module the vault names this engine's login values under: own/<node>/mesh-host/<login> (ADR 0583 §3,
|
||||
// insight).
|
||||
const loginSecretModule = "mesh-host"
|
||||
|
||||
// fetchLogin asks the vault, through its existing hand-out of a machine's own secrets, for the value of login, sealed to this
|
||||
// node's key, and opens it with the sealing key; no new verb is asked for it.
|
||||
func fetchLogin(queue *link.Session, statePath string) func(context.Context, string) (string, error) {
|
||||
unseal := sealOpener(statePath)
|
||||
return func(ctx context.Context, login string) (string, error) {
|
||||
// The vault names the login values under the engine's own module, which is §527's (§634).
|
||||
own, err := queue.Own.Wait(ctx)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("the node-engine's own module is not named yet, so the vault's name of its login value is not known: %w", err)
|
||||
}
|
||||
raw, err := queue.AskVault(ctx, "own", map[string]any{})
|
||||
if err != nil {
|
||||
return "", err
|
||||
@@ -111,7 +112,7 @@ func fetchLogin(queue *link.Session, statePath string) func(context.Context, str
|
||||
return "", fmt.Errorf("the vault's own answer is not one: %w", err)
|
||||
}
|
||||
for _, o := range a.Own {
|
||||
if o.Module != loginSecretModule || o.Name != login {
|
||||
if o.Module != own || o.Name != login {
|
||||
continue
|
||||
}
|
||||
sealed := string(o.Sealed)
|
||||
@@ -125,7 +126,7 @@ func fetchLogin(queue *link.Session, statePath string) func(context.Context, str
|
||||
}
|
||||
return string(value), nil
|
||||
}
|
||||
return "", fmt.Errorf("the vault hands this node no value for %s (own/%s/%s/%s)", login, queueNode(queue), loginSecretModule, login)
|
||||
return "", fmt.Errorf("the vault hands this node no value for %s (own/%s/%s/%s)", login, queueNode(queue), own, login)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -36,6 +36,7 @@ import (
|
||||
"github.com/novox/mesh-host/internal/agent"
|
||||
"github.com/novox/mesh-host/internal/apply"
|
||||
"github.com/novox/mesh-host/internal/bundle"
|
||||
"github.com/novox/mesh-host/internal/compositions"
|
||||
"github.com/novox/mesh-host/internal/firewall"
|
||||
"github.com/novox/mesh-host/internal/identity"
|
||||
"github.com/novox/mesh-host/internal/inventory"
|
||||
@@ -987,7 +988,12 @@ func runLink(ctx context.Context, opts options) error {
|
||||
// What this engine trusts a mesh signature from: the vault's key alone, once read from its signers (novox/hq ADR 0550
|
||||
// §2; issue 1126: the controller's key never, the controller ended by its unassignment).
|
||||
trust := newMeshTrust()
|
||||
own := link.NewOwnName()
|
||||
// The engine's own module, first from its applied record of the build whose fingerprint is its own (design 47 G §527,
|
||||
// §634, A-149); on a first start with no such record, from its first composing.
|
||||
own.Set(ownModuleApplied(appliedStoreFor(opts.state), os.Getenv(compositions.FingerprintVar)))
|
||||
queue := &link.Session{
|
||||
Own: own,
|
||||
Membership: membership,
|
||||
// And the public halves and certificate its record says (novox/hq ADR 0555 §3), read at each facts report.
|
||||
Keys: recordKeys(opts.state, trust.Trusts),
|
||||
|
||||
@@ -601,16 +601,13 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
|
||||
// And the seat list it reads through the seat mesh-catalog, from its bootstrapped definitions until it has: the
|
||||
// composer hands nothing new meanwhile, and the node says it has waited five minutes (novox/hq ADR 0636 §1).
|
||||
// The finding writer is started first, so it holds what an earlier run left before the reader writes or clears any.
|
||||
findings, ferr := newFindingsOfTheNode(session, node, say)
|
||||
if ferr != nil { // a writer that cannot be made is a fault of its configuration: loud, never a node without findings
|
||||
say("the node-engine cannot make its finding writer, so its node agent stops: " + ferr.Error())
|
||||
stopRun()
|
||||
cancel()
|
||||
return
|
||||
}
|
||||
if werr := findings.Start(run); werr != nil {
|
||||
say("the node-engine's findings are kept in its state and not said as events until it is started again: " + werr.Error())
|
||||
if session.Own == nil {
|
||||
session.Own = link.NewOwnName()
|
||||
}
|
||||
findings := &lateFindings{own: session.Own, make: func(module string) (*findingsOfTheNode, error) {
|
||||
return newFindingsOfTheNode(session, node, module, say)
|
||||
}}
|
||||
findings.Start(run, say)
|
||||
seats, err := startSeatList(run, session, node, findings, say)
|
||||
if err != nil {
|
||||
// No reader is no gate that lets everything through: the node acts on nothing new, as one that waits does.
|
||||
@@ -618,10 +615,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
|
||||
}
|
||||
// The marks of the haves run off the composition pass, each bounded (see marker).
|
||||
marks := newMarker(run)
|
||||
heldFindings := &seatwait.HeldKeeper{Node: node, Observer: findingsModule, Writer: findings, Ctx: run, Say: say}
|
||||
heldFindings := &seatwait.HeldKeeper{Node: node, ObserverOf: findings.observer, Writer: findings, Ctx: run, Say: say}
|
||||
hooks := composerHooks{
|
||||
entryNotFound: func(fp string, notFound bool) {
|
||||
if err := seatwait.SayEntryNotFound(run, findings, node, findingsModule, fp, notFound, time.Now); err != nil {
|
||||
if err := seatwait.SayEntryNotFound(run, findings, node, findings.observer(), fp, notFound, time.Now); err != nil {
|
||||
say("the finding that the catalog half holds no entry of " + fp + " cannot be kept: " + err.Error())
|
||||
}
|
||||
},
|
||||
@@ -687,6 +684,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
|
||||
// composition bucket) is never applied. While the seat's assignment holds no key for this node it composes
|
||||
// nothing and keeps running what it runs; a assignment it cannot watch stops the agent, said, and it begins again.
|
||||
composer := selfComposer(session, node, hooks, say)
|
||||
composer.OnEngine = session.Own.Set // the first composing names the engine's own module on a first start (§527, §634)
|
||||
machinesNow.Store(composer) // the machines a failing network part points at, by the assignments (issue 774)
|
||||
composerNow.Store(composer) // what the verb compose asks a pass of, this run's alone
|
||||
session.SetManifestOf(composer.ManifestOf) // the event running names the manifest of the build that runs
|
||||
|
||||
@@ -231,10 +231,19 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed {
|
||||
return b
|
||||
}
|
||||
|
||||
// itOwnModule is the node-engine's own module of the bed: what its applied record of its own build says (§527, §634).
|
||||
const itOwnModule = "mesh-host"
|
||||
|
||||
func itOwn() *link.OwnName {
|
||||
o := link.NewOwnName()
|
||||
o.Set(itOwnModule)
|
||||
return o
|
||||
}
|
||||
|
||||
// start runs the node agent under a session on the bus.
|
||||
func (b *itBed) start() (stop func()) {
|
||||
b.t.Helper()
|
||||
b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
||||
b.session = &link.Session{Own: itOwn(), Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
||||
Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
|
||||
b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode})
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
@@ -1055,7 +1064,7 @@ func (c *fiveMinutes) pass() {
|
||||
|
||||
// findingSubject is the subject of one of the engine's finding events on node (views.FindingSubjects).
|
||||
func findingSubject(event, node string) string {
|
||||
return "mesh.stream." + findingsModule + ".event." + event + "." + node
|
||||
return "mesh.stream." + itOwnModule + ".event." + event + "." + node
|
||||
}
|
||||
|
||||
// ADR 0636 §1, §8, whole against a real bus and the stand-in of the seat: while the seat refuses, the node composes and
|
||||
|
||||
@@ -0,0 +1,56 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions"
|
||||
"github.com/novox/mesh-host/internal/link"
|
||||
)
|
||||
|
||||
// Design 47 G §527, §634, A-149: the node-engine's own module is the module of its applied record at its own fingerprint,
|
||||
// and its finding writer is made, observing as that module, once the name is known: at once from the record, else when the
|
||||
// first composing names it. No name is fixed in code.
|
||||
func TestTheEnginesOwnNameIsReadFromItsAppliedRecordOrItsFirstComposing(t *testing.T) {
|
||||
fp := "src2:" + strings.Repeat("a", 64)
|
||||
applied := appliedStoreFor(filepath.Join(t.TempDir(), "state", "state.json"))
|
||||
for module, f := range map[string]string{"node": "", "the-engine": fp, "other": "src2:" + strings.Repeat("b", 64)} {
|
||||
if err := applied.Keep(module, []byte(`{"module":"`+module+`","fp":"`+f+`"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, tc := range []struct{ fp, want string }{{fp, "the-engine"}, {"", ""}, {"src2:" + strings.Repeat("c", 64), ""}} {
|
||||
if got := ownModuleApplied(applied, tc.fp); got != tc.want {
|
||||
t.Errorf("the own module at %q is %q, want %q", tc.fp, got, tc.want)
|
||||
}
|
||||
}
|
||||
|
||||
own := link.NewOwnName()
|
||||
made := make(chan string, 1)
|
||||
late := &lateFindings{own: own, make: func(module string) (*findingsOfTheNode, error) {
|
||||
made <- module
|
||||
return nil, context.Canceled // the writer itself is the SDK's: only when it is made is tested here
|
||||
}}
|
||||
ctx, stop := context.WithCancel(context.Background())
|
||||
defer stop()
|
||||
var said []string
|
||||
late.Start(ctx, func(s string) { said = append(said, s) })
|
||||
if err := late.Write(ctx, condseat.Finding{}); err == nil || late.Held() != nil {
|
||||
t.Fatalf("a finding was taken before the engine's own module was known: %v", err)
|
||||
}
|
||||
if len(made) != 0 {
|
||||
t.Fatal("the writer was made before a name")
|
||||
}
|
||||
own.Set("the-engine") // the first composing names it
|
||||
select {
|
||||
case m := <-made:
|
||||
if m != "the-engine" || late.observer() != "the-engine" {
|
||||
t.Fatalf("the writer was made as %q, observing as %q", m, late.observer())
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the writer was never made after the name was known")
|
||||
}
|
||||
}
|
||||
@@ -135,7 +135,7 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T)
|
||||
|
||||
// startSession is a session on the node's own connection, with no node agent run on it.
|
||||
func (b *itBed) startSession() *link.Session {
|
||||
b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
||||
b.session = &link.Session{Own: itOwn(), Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
|
||||
Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
|
||||
b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode})
|
||||
return b.session
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@@ -102,10 +103,6 @@ func (realClock) AfterFunc(d time.Duration, f func()) seatwait.Timer {
|
||||
return time.AfterFunc(d, f)
|
||||
}
|
||||
|
||||
// findingsModule is the module whose own stream the node-engine says its findings on, and whose name its findings carry as
|
||||
// their observer (module.json's `module` and its stream `mesh-host-findings`, design 47 G §383).
|
||||
const findingsModule = "mesh-host"
|
||||
|
||||
// nodeFindings is the node-engine's finding writer: mesh-sdk's held writer, for the engine's findings are raised as they
|
||||
// happen and not judged afresh (design 47 G A-29, A-40), with the contract of the seat mesh-conditions.
|
||||
type nodeFindings = views.HeldWriter[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]
|
||||
@@ -120,9 +117,9 @@ type findingsOfTheNode struct {
|
||||
|
||||
// newFindingsOfTheNode makes the writer at once; it reads the state when it is started (A-40). Its events are published on
|
||||
// the engine's own stream awaiting the bus's acknowledgement and its consumer of `findings-asked` is its own.
|
||||
func newFindingsOfTheNode(session *link.Session, node string, say func(string)) (*findingsOfTheNode, error) {
|
||||
func newFindingsOfTheNode(session *link.Session, node, module string, say func(string)) (*findingsOfTheNode, error) {
|
||||
writer, err := views.NewHeldWriter(views.WriterConfig[condseat.Finding, condseat.FindingCleared, condseat.FindingsStated, condseat.FindingsAskedEvent]{
|
||||
Module: findingsModule,
|
||||
Module: module,
|
||||
Node: node,
|
||||
State: link.FindingsState{Session: session},
|
||||
Kinds: []string{seatwait.Kind, seatwait.HeldKind, seatwait.EntryNotFoundKind},
|
||||
@@ -148,7 +145,7 @@ func (w *findingsOfTheNode) Start(ctx context.Context) error {
|
||||
// startSeatList starts this run's reader: its bootstrapped seats are given to catalogue at once, and every change of what
|
||||
// it holds again, saying so to the composer and to the waiter that writes the five-minute finding, and stating again what
|
||||
// the engine reads (link.Session.RestateReads, ADR 0618 §4): the seats it states are those of the list it holds.
|
||||
func startSeatList(ctx context.Context, session *link.Session, node string, findings seatwait.Writer, say func(string)) (*seatListState, error) {
|
||||
func startSeatList(ctx context.Context, session *link.Session, node string, findings *lateFindings, say func(string)) (*seatListState, error) {
|
||||
st := &seatListState{changed: make(chan struct{}, 1)}
|
||||
// Start returns at once and reads the seat on a goroutine, which may announce a change before the waiter exists.
|
||||
var announce atomic.Pointer[func()]
|
||||
@@ -166,7 +163,7 @@ func startSeatList(ctx context.Context, session *link.Session, node string, find
|
||||
return nil, fmt.Errorf("the node-engine cannot follow the seat list: %w", err)
|
||||
}
|
||||
st.reader = reader
|
||||
st.waiter = &seatwait.Waiter{Node: node, Observer: findingsModule, Reader: reader, Writer: findings, Ctx: ctx,
|
||||
st.waiter = &seatwait.Waiter{Node: node, ObserverOf: findings.observer, Reader: reader, Writer: findings, Ctx: ctx,
|
||||
Clock: seatWaitClock, Say: say}
|
||||
told := func() {
|
||||
st.waiter.Changed()
|
||||
@@ -194,3 +191,71 @@ func startSeatList(ctx context.Context, session *link.Session, node string, find
|
||||
seatListNow.Store(st)
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// lateFindings is the node-engine's finding writer, made when the engine's own module is known (design 47 G §527, §634,
|
||||
// A-149): at once from its applied record, or, on a first start with no such record, when the first composing names it. Till
|
||||
// then a write or a clear is refused in words, never lost in silence.
|
||||
type lateFindings struct {
|
||||
own *link.OwnName
|
||||
make func(module string) (*findingsOfTheNode, error)
|
||||
mu sync.Mutex
|
||||
w *findingsOfTheNode
|
||||
}
|
||||
|
||||
// observer is the module the findings are observed as: the engine's own, "" while it is not known.
|
||||
func (f *lateFindings) observer() string { return f.own.Get() }
|
||||
|
||||
// Start makes the writer and starts it as soon as the own name is known: in this call when it already is, else on a goroutine.
|
||||
func (f *lateFindings) Start(ctx context.Context, say func(string)) {
|
||||
begin := func(module string) {
|
||||
w, err := f.make(module)
|
||||
if err != nil { // a writer that cannot be made is a fault of its configuration: loud
|
||||
say("the node-engine cannot make its finding writer: " + err.Error())
|
||||
return
|
||||
}
|
||||
if err := w.Start(ctx); err != nil {
|
||||
say("the node-engine's findings are kept in its state and not said as events until it is started again: " + err.Error())
|
||||
}
|
||||
f.mu.Lock()
|
||||
f.w = w
|
||||
f.mu.Unlock()
|
||||
}
|
||||
if name := f.own.Get(); name != "" {
|
||||
begin(name)
|
||||
return
|
||||
}
|
||||
go func() {
|
||||
if name, err := f.own.Wait(ctx); err == nil {
|
||||
begin(name)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
var errOwnNameUnknown = errors.New("the node-engine's own module is not yet named by its applied record or its first composing, so its finding writer is not made")
|
||||
|
||||
func (f *lateFindings) writer() *findingsOfTheNode {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return f.w
|
||||
}
|
||||
|
||||
func (f *lateFindings) Write(ctx context.Context, fi condseat.Finding) error {
|
||||
if w := f.writer(); w != nil {
|
||||
return w.Write(ctx, fi)
|
||||
}
|
||||
return errOwnNameUnknown
|
||||
}
|
||||
|
||||
func (f *lateFindings) Clear(ctx context.Context, scope, id, kind, why string) error {
|
||||
if w := f.writer(); w != nil {
|
||||
return w.Clear(ctx, scope, id, kind, why)
|
||||
}
|
||||
return errOwnNameUnknown
|
||||
}
|
||||
|
||||
func (f *lateFindings) Held() []condseat.Finding {
|
||||
if w := f.writer(); w != nil {
|
||||
return w.Held()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -131,6 +131,28 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
||||
if handle == nil || l.conn == nil {
|
||||
return
|
||||
}
|
||||
// Discovery names the module whose code answers: the engine's own, read from §527 and no fixed name (§634). Known at
|
||||
// once from the applied record; on a first start it is served once the first composing names it.
|
||||
if l.own == nil || l.own.Get() != "" {
|
||||
l.serveEngineSeat(handle, l.own.Get())
|
||||
return
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
l.engineMu.Lock()
|
||||
l.stopWait = cancel
|
||||
l.engineMu.Unlock()
|
||||
go func() {
|
||||
if name, err := l.own.Wait(ctx); err == nil {
|
||||
l.serveEngineSeat(handle, name)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// UseOwnName gives the link the engine's own name its discovery reads; before ServeEngineSeat.
|
||||
func (l *natsLink) UseOwnName(o *OwnName) { l.own = o }
|
||||
|
||||
// serveEngineSeat serves the seat as the module module.
|
||||
func (l *natsLink) serveEngineSeat(handle EngineFunc, module string) {
|
||||
svc, err := micro.AddService(l.conn, micro.Config{Name: EngineSeat, Version: "1.0.0",
|
||||
Description: "the node seat node-engine of " + l.node + " (novox/hq ADR 0563)", QueueGroupDisabled: true,
|
||||
Metadata: map[string]string{"seat": EngineSeat, "scope": "node", "node": l.node}})
|
||||
@@ -141,13 +163,15 @@ func (l *natsLink) ServeEngineSeat(handle EngineFunc) {
|
||||
for _, v := range ServedEngineSeat().Serves {
|
||||
verb := v.Name
|
||||
answer := EngineAnswer(handle, verb, l.node, apart)
|
||||
meta := EngineEndpointMetadata(v, l.node)
|
||||
meta := EngineEndpointMetadata(v, l.node, module)
|
||||
_ = svc.AddEndpoint(verb, answer, micro.WithEndpointSubject(EngineToolSubject(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
_ = svc.AddEndpoint(verb+"-called", answer, micro.WithEndpointSubject(EngineCallPattern(verb, l.node)),
|
||||
micro.WithEndpointMetadata(meta))
|
||||
}
|
||||
l.engineMu.Lock()
|
||||
l.engine = svc
|
||||
l.engineMu.Unlock()
|
||||
}
|
||||
|
||||
// answeredApart are the verbs answered off the subscription that delivered them: each endpoint's subscription hands its
|
||||
@@ -184,20 +208,17 @@ func EngineAnswer(handle EngineFunc, verb, node string, apart chan struct{}) mic
|
||||
}
|
||||
}
|
||||
|
||||
// EngineModule is the module whose code answers the seat, as discovery names it: the node-engine's manifest's.
|
||||
const EngineModule = "mesh-host"
|
||||
|
||||
// EngineEndpointMetadata is what discovery reads of one verb's endpoint, in the shape every tool runner and the
|
||||
// controller announce a seat's verb in (mesh-tools node-tools' announce package): its kind, the seat, its scope, the
|
||||
// module whose code answers, the verb, the node, its description and its schema. Without the kind and the scope the
|
||||
// console listed none of it, and `<node>/node-engine.<verb>` answered that nothing in the mesh is called node-engine
|
||||
// (the sweep of 2026-10-30).
|
||||
func EngineEndpointMetadata(v catalogue.Verb, node string) map[string]string {
|
||||
func EngineEndpointMetadata(v catalogue.Verb, node, module string) map[string]string {
|
||||
schema, err := json.Marshal(v.Input)
|
||||
if err != nil || v.Input == nil {
|
||||
schema = []byte("{}")
|
||||
}
|
||||
return map[string]string{"kind": "seat", "seat": EngineSeat, "scope": "node", "module": EngineModule,
|
||||
return map[string]string{"kind": "seat", "seat": EngineSeat, "scope": "node", "module": module,
|
||||
"tool": v.Name, "node": node, "interchangeable": "false", "description": v.Description, "schema": string(schema)}
|
||||
}
|
||||
|
||||
|
||||
@@ -8,6 +8,7 @@ import (
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
@@ -48,6 +49,10 @@ type natsLink struct {
|
||||
moving *nats.Subscription
|
||||
// engine serves the node seat node-engine, and warrants hears the operator's answers to its asks (ADR 0563).
|
||||
engine micro.Service
|
||||
engineMu sync.Mutex
|
||||
// own is the engine's own module, which discovery names (§634); stopWait ends a wait for it when the link closes.
|
||||
own *OwnName
|
||||
stopWait context.CancelFunc
|
||||
warrants *nats.Subscription
|
||||
node string
|
||||
// user is the bus user this link authenticated as (node.<n> or node.<n>--b, ADR 0583 §1).
|
||||
@@ -295,9 +300,14 @@ func (l *natsLink) Close() {
|
||||
if l.judging != nil {
|
||||
_ = l.judging.Unsubscribe()
|
||||
}
|
||||
l.engineMu.Lock()
|
||||
if l.stopWait != nil {
|
||||
l.stopWait()
|
||||
}
|
||||
if l.engine != nil {
|
||||
_ = l.engine.Stop()
|
||||
}
|
||||
l.engineMu.Unlock()
|
||||
if l.warrants != nil {
|
||||
_ = l.warrants.Unsubscribe()
|
||||
}
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// OwnName is the module that is this node-engine: the module whose catalog entry claims the seat node-engine, which the
|
||||
// engine learns first from its applied record of the build whose fingerprint is its own and, on a first start with no such
|
||||
// record, from its first composing (design 47 G §527, §634, A-149). Its discovery, its finding writer and its login names
|
||||
// in the vault read it here and fix none.
|
||||
type OwnName struct {
|
||||
mu sync.Mutex
|
||||
name string
|
||||
ready chan struct{}
|
||||
}
|
||||
|
||||
// NewOwnName makes an own name not yet known.
|
||||
func NewOwnName() *OwnName { return &OwnName{ready: make(chan struct{})} }
|
||||
|
||||
// Set keeps the name; an empty one changes nothing. The first name opens Wait for everyone.
|
||||
func (o *OwnName) Set(name string) {
|
||||
if o == nil || name == "" {
|
||||
return
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
if o.name == "" {
|
||||
close(o.ready)
|
||||
}
|
||||
o.name = name
|
||||
}
|
||||
|
||||
// Get is the name, "" while it is not known.
|
||||
func (o *OwnName) Get() string {
|
||||
if o == nil {
|
||||
return ""
|
||||
}
|
||||
o.mu.Lock()
|
||||
defer o.mu.Unlock()
|
||||
return o.name
|
||||
}
|
||||
|
||||
// Wait returns the name once it is known, or ctx's error.
|
||||
func (o *OwnName) Wait(ctx context.Context) (string, error) {
|
||||
if o == nil {
|
||||
return "", errors.New("this session names no own module")
|
||||
}
|
||||
select {
|
||||
case <-o.ready:
|
||||
return o.Get(), nil
|
||||
case <-ctx.Done():
|
||||
return "", ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
// OwnNamed is a link that reads the node-engine's own module from an OwnName when it serves its seat.
|
||||
type OwnNamed interface {
|
||||
UseOwnName(*OwnName)
|
||||
}
|
||||
@@ -0,0 +1,32 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
// Design 47 G §527, §634: discovery names the module the engine reads of itself, Wait holds until it is known, and an
|
||||
// empty name changes nothing.
|
||||
func TestTheOwnNameIsWaitedForAndDiscoveryNamesIt(t *testing.T) {
|
||||
o := NewOwnName()
|
||||
o.Set("")
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond)
|
||||
defer cancel()
|
||||
if _, err := o.Wait(ctx); err == nil || o.Get() != "" {
|
||||
t.Fatalf("an unnamed engine's wait ended with %v, name %q", err, o.Get())
|
||||
}
|
||||
o.Set("the-engine")
|
||||
if got, err := o.Wait(context.Background()); err != nil || got != "the-engine" {
|
||||
t.Fatalf("waited %q, %v", got, err)
|
||||
}
|
||||
if got := EngineEndpointMetadata(catalogue.Verb{Name: "composition"}, "n1", o.Get())["module"]; got != "the-engine" {
|
||||
t.Fatalf("discovery names the module %q", got)
|
||||
}
|
||||
var none *OwnName
|
||||
if _, err := none.Wait(context.Background()); err == nil || none.Get() != "" {
|
||||
t.Fatalf("a session naming none waited: %v", err)
|
||||
}
|
||||
}
|
||||
@@ -217,6 +217,9 @@ func serve(ctx context.Context, link Link, m Membership, queue *Session, say Ann
|
||||
// link, and what was answered while no link was open caught up from the router's records.
|
||||
if queue.Acts != nil {
|
||||
if s, ok := link.(EngineSeats); ok {
|
||||
if o, named := link.(OwnNamed); named {
|
||||
o.UseOwnName(queue.Own)
|
||||
}
|
||||
s.ServeEngineSeat(queue.Acts.Call)
|
||||
// And what it reads of a manifest, said once for this link (novox/hq ADR 0618 §4).
|
||||
queue.SayReads()
|
||||
|
||||
@@ -26,6 +26,9 @@ import (
|
||||
// **It applies nothing and orders nothing.** A module is applied by the node agent from its want. Nothing here
|
||||
// receives, orders or reconciles a whole declaration.
|
||||
type Session struct {
|
||||
// Own is the module that is this node-engine (§527, §634): its discovery, finding writer and vault login names read it.
|
||||
// Nil names none, as a test's session does.
|
||||
Own *OwnName
|
||||
// Fallback is the engine's fall back to its old login its health report carries now (Rotator.Fallback); nil says none.
|
||||
Fallback func() *nodeengine.LoginFallback
|
||||
// linkedUser is the bus user the link open now authenticated as, guarded by mu (WhenLinkedAs).
|
||||
|
||||
@@ -25,7 +25,9 @@ const HeldKind = "module-held"
|
||||
type HeldKeeper struct {
|
||||
Node string
|
||||
Observer string
|
||||
Writer Writer
|
||||
// ObserverOf, when set, answers the observer at each write (see Waiter).
|
||||
ObserverOf func() string
|
||||
Writer Writer
|
||||
// Ctx is the run's context for the writer's acts; nil is the background.
|
||||
Ctx context.Context
|
||||
// Now is the clock; nil is time.Now.
|
||||
@@ -78,7 +80,7 @@ func (k *HeldKeeper) Reconcile(held map[string]string) {
|
||||
continue
|
||||
}
|
||||
id := HeldID(m, k.Node)
|
||||
f := condseat.Finding{Observer: k.Observer, Kind: HeldKind, Scope: scopeModule, ID: id, Severity: condseat.FindingSeverityWarning,
|
||||
f := condseat.Finding{Observer: observerOf(k.Observer, k.ObserverOf), Kind: HeldKind, Scope: scopeModule, ID: id, Severity: condseat.FindingSeverityWarning,
|
||||
Source: HeldKind,
|
||||
Subject: condseat.Subject{Scope: scopeModule, ID: id, Machine: k.Node},
|
||||
Summary: fmt.Sprintf("%s is held on %s: %s", m, k.Node, why),
|
||||
|
||||
@@ -65,9 +65,12 @@ func stands(w Writer, scope, id, kind string) bool {
|
||||
type Waiter struct {
|
||||
Node string
|
||||
Observer string
|
||||
Reader Reader
|
||||
Writer Writer
|
||||
Clock Clock
|
||||
// ObserverOf, when set, answers the observer at each write: the node-engine's own module, which a first start learns
|
||||
// from its first composing (design 47 G §527, §634).
|
||||
ObserverOf func() string
|
||||
Reader Reader
|
||||
Writer Writer
|
||||
Clock Clock
|
||||
// Ctx is the run's context for the writer's acts; nil is the background.
|
||||
Ctx context.Context
|
||||
// Say tells the node's journal; nil says nothing.
|
||||
@@ -143,7 +146,7 @@ func (w *Waiter) expired() {
|
||||
if w.stands() {
|
||||
return
|
||||
}
|
||||
f := condseat.Finding{Observer: w.Observer, Kind: Kind, Scope: scopeMachine, ID: w.Node, Severity: condseat.FindingSeverityWarning,
|
||||
f := condseat.Finding{Observer: observerOf(w.Observer, w.ObserverOf), Kind: Kind, Scope: scopeMachine, ID: w.Node, Severity: condseat.FindingSeverityWarning,
|
||||
Source: Kind, Subject: condseat.Subject{Scope: scopeMachine, ID: w.Node, Machine: w.Node},
|
||||
Summary: fmt.Sprintf("%s has read no complete seat list since %s: %s", w.Node, w.since.UTC().Format(time.RFC3339), st.Health()),
|
||||
Headline: fmt.Sprintf("%s waits for the seat list", w.Node),
|
||||
@@ -161,3 +164,11 @@ func (w *Waiter) say(s string) {
|
||||
w.Say(s)
|
||||
}
|
||||
}
|
||||
|
||||
// observerOf is of() when it is given, else fixed.
|
||||
func observerOf(fixed string, of func() string) string {
|
||||
if of != nil {
|
||||
return of()
|
||||
}
|
||||
return fixed
|
||||
}
|
||||
|
||||
@@ -160,7 +160,10 @@ const VaultSecretChanged = "mesh.mod.mesh-vault.event.secret-changed"
|
||||
|
||||
// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed).
|
||||
type Composer struct {
|
||||
env Env
|
||||
// OnEngine is told the node-engine's own module each time a composing finds it (design 47 G §527, §634); set before the
|
||||
// composer runs.
|
||||
OnEngine func(module string)
|
||||
env Env
|
||||
|
||||
mu sync.Mutex
|
||||
wants map[string]meshwants.WantsByWantNodeModuleValue // by module, the node's own under agent.NodeModule
|
||||
@@ -2453,7 +2456,11 @@ func (c *Composer) noteEngine(module string) {
|
||||
}
|
||||
c.mu.Lock()
|
||||
c.engine = module
|
||||
on := c.OnEngine
|
||||
c.mu.Unlock()
|
||||
if on != nil {
|
||||
on(module)
|
||||
}
|
||||
}
|
||||
|
||||
// EngineModule is the node-engine's own module (design 47 G §527), for the apply guard; "" until a composing found it.
|
||||
|
||||
Reference in New Issue
Block a user