ENG-2: the node's record says the capabilities its tool runner states of its calls, asked of the runner's PING at each report (issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -157,6 +157,18 @@ func TestADirectlyAssignedModuleTheMachineCannotHostIsReportedNotRefused(t *test
|
||||
if !strings.Contains(WrongMachine("xorg", "seat", "server"), "wrong machine") {
|
||||
t.Errorf("the report reads like a missing module")
|
||||
}
|
||||
|
||||
// And a module whose calls only a tool runner stating them answers (ENG-2, issue 1966) is held off a machine whose record
|
||||
// does not say them, naming the capability; on one that does, it is hosted.
|
||||
withCalls := mcat.Node{Name: "server", Capabilities: map[string]bool{"container-runtime": true, "ledger-calls": true}}
|
||||
ledgerOwner := shelf(mod("records", nil, nil, []string{"ledger-calls"}))
|
||||
held, err := Resolve(ledgerOwner, []string{"records"}, server, mcat.World{})
|
||||
if err != nil || len(held.Unhostable) != 1 || len(held.Unhostable[0].Missing) != 1 || held.Unhostable[0].Missing[0] != "ledger-calls" {
|
||||
t.Errorf("a ledger owner on a machine whose runner states no ledger calls: %+v, %v", held.Unhostable, err)
|
||||
}
|
||||
if got, err := Resolve(ledgerOwner, []string{"records"}, withCalls, mcat.World{}); err != nil || len(got.Unhostable) != 0 || len(got.Modules) != 1 {
|
||||
t.Errorf("a ledger owner on a machine whose runner states them was held: %v %v", got.Unhostable, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAnUnhostableModuleSomethingRequiresRefusesTheNode(t *testing.T) {
|
||||
|
||||
@@ -296,7 +296,9 @@ func (q *Session) noteApply(r Report) bool {
|
||||
func (q *Session) nodeFactsWithRecord(r Report, before *delivery.HaveNode) *delivery.HaveNode {
|
||||
n := nodeFacts(r)
|
||||
if q.Keys != nil {
|
||||
n.Record = recordOf(q.Membership.Node, r, q.Keys())
|
||||
keys := q.Keys()
|
||||
keys.Stated = q.runnerStated
|
||||
n.Record = recordOf(q.Membership.Node, r, keys)
|
||||
}
|
||||
if before != nil && before.Record != nil {
|
||||
switch {
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -31,6 +32,15 @@ type haveLink struct {
|
||||
bindFails, bindStuck bool
|
||||
// bindNoResponders fails that many binds with the bus's no responders, as while it reloads its accounts.
|
||||
bindNoResponders int
|
||||
// stated is what the tool runner's PING says of itself (Stater); statedErr is the runner not answering.
|
||||
stated map[string]string
|
||||
statedErr error
|
||||
}
|
||||
|
||||
func (l *haveLink) Stated(context.Context, string, string) (map[string]string, error) {
|
||||
l.hmu.Lock()
|
||||
defer l.hmu.Unlock()
|
||||
return l.stated, l.statedErr
|
||||
}
|
||||
|
||||
type haveEntry struct {
|
||||
@@ -259,6 +269,7 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
keys := RecordKeys{TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk", Certificate: "cert",
|
||||
Trusts: []string{"controller key sha256:aa", "vault key sha256:bb"}} // and what it trusts (L3 s6, ADR 0550 §2)
|
||||
q.Keys = func() RecordKeys { return keys }
|
||||
l.stated = map[string]string{CapabilitiesKey: "ledger-calls, bus-state-calls"} // the tool runner's calls (ENG-2)
|
||||
rep.Profile = map[string]any{"capabilities": []any{map[string]any{"name": "systemd", "present": true},
|
||||
map[string]any{"name": "docker", "present": false}, map[string]any{"name": "apt", "present": true}},
|
||||
"machine": map[string]any{"networks": []any{map[string]any{"link": "eth0", "network": "192.0.2.0/24"}}}}
|
||||
@@ -269,7 +280,7 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
t.Fatalf("the node key holds no record: %+v, %v", got, err)
|
||||
}
|
||||
wantRecord := delivery.NodeRecord{V: delivery.Version, Node: "n1", TunnelKey: "tk", SealingKey: "sk", ServingKey: "pk",
|
||||
Certificate: "cert", Trusts: keys.Trusts, Capabilities: []string{"apt", "systemd"}, OutwardLinks: []string{"eth0"}, NetworksSaid: true,
|
||||
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,
|
||||
ManifestReads: catalogue.ManifestReads()} // what it reads of a manifest: its contract's (ADR 0614 §4)
|
||||
if !delivery.SameNodeRecord(*got.Record, wantRecord) {
|
||||
@@ -280,6 +291,22 @@ func TestTheNodeKeyCarriesTheNodesFacts(t *testing.T) {
|
||||
if l.kv["n1.node"].rev != at {
|
||||
t.Errorf("a report changing nothing wrote the node key again")
|
||||
}
|
||||
// A runner that does not answer states what it stated before; one answering without any states none (an older runner).
|
||||
l.statedErr = fmt.Errorf("%w: nothing answered", ErrNoAnswer)
|
||||
l.stated = nil
|
||||
q.SayFacts(context.Background(), rep)
|
||||
if l.kv["n1.node"].rev != at {
|
||||
t.Errorf("a runner that did not answer changed the record")
|
||||
}
|
||||
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"}) {
|
||||
t.Errorf("a runner stating no calls: revision %d (was %d), capabilities %v", l.kv["n1.node"].rev, at, got.Record.Capabilities)
|
||||
}
|
||||
at++
|
||||
l.stated = map[string]string{CapabilitiesKey: "ledger-calls,bus-state-calls"}
|
||||
q.SayFacts(context.Background(), rep)
|
||||
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" {
|
||||
|
||||
@@ -2,6 +2,7 @@ package link
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"slices"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
@@ -26,6 +27,9 @@ type RecordKeys struct {
|
||||
Authority string
|
||||
// Trusts is what this machine trusts a mesh signature from now (delivery.Trusts, ADR 0550 §2).
|
||||
Trusts []string
|
||||
// Stated is the capabilities its tool runner states of the calls it answers (CapabilitiesKey), which the record says
|
||||
// beside the machine's own: a module whose manifest asks for one is held off a machine whose runner does not state it.
|
||||
Stated []string
|
||||
}
|
||||
|
||||
// manifestReads is what this engine reads of a manifest (novox/hq ADR 0614 §4): the reader capabilities of the mesh-sdk it
|
||||
@@ -60,6 +64,11 @@ func recordOf(node string, r Report, keys RecordKeys) *delivery.NodeRecord {
|
||||
rec.Capabilities = append(rec.Capabilities, c.Name)
|
||||
}
|
||||
}
|
||||
for _, c := range keys.Stated {
|
||||
if c != "" && !slices.Contains(rec.Capabilities, c) {
|
||||
rec.Capabilities = append(rec.Capabilities, c)
|
||||
}
|
||||
}
|
||||
sort.Strings(rec.Capabilities)
|
||||
if p.Machine != nil {
|
||||
rec.NetworksSaid = true
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -79,6 +80,10 @@ type Session struct {
|
||||
haveAgainPending bool
|
||||
haveAgainTries int
|
||||
|
||||
// runnerStated is the capabilities this machine's tool runner last stated of the calls it answers (askRunner), said in the
|
||||
// node's record. Held under saying.
|
||||
runnerStated []string
|
||||
|
||||
// saying is held while a facts event is noted, so the haves are written in the order the facts are made.
|
||||
saying sync.Mutex
|
||||
|
||||
@@ -161,6 +166,7 @@ func (q *Session) SayFacts(ctx context.Context, r Report) bool {
|
||||
return false
|
||||
}
|
||||
r.Node = q.Membership.Node
|
||||
q.askRunner(ctx, bus)
|
||||
if q.noteApply(r) {
|
||||
q.writeHaves(ctx, bus)
|
||||
}
|
||||
@@ -404,3 +410,30 @@ func (q *Session) operatorAsks() (OperatorAsks, error) {
|
||||
}
|
||||
return asking, nil
|
||||
}
|
||||
|
||||
// runnerAsk bounds the question to the tool runner.
|
||||
const runnerAsk = 3 * time.Second
|
||||
|
||||
// askRunner reads what this machine's tool runner states of the calls it answers, asked of its own instance with the PING
|
||||
// the witness asks (ENG-2, issue 1966), at each report: a runner replaced by a build that states more, or less, is said in
|
||||
// the record from the next report. A runner that does not answer states what it stated before, since its absence says
|
||||
// nothing of its calls; one that answers without stating any states none, as an older runner does. Called under saying.
|
||||
func (q *Session) askRunner(ctx context.Context, bus Bus) {
|
||||
stater, ok := bus.(Stater)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
asking, cancel := context.WithTimeout(ctx, runnerAsk)
|
||||
defer cancel()
|
||||
md, err := stater.Stated(asking, ToolRunnerService, q.Membership.Node)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
var stated []string
|
||||
for _, c := range strings.Split(md[CapabilitiesKey], ",") {
|
||||
if c = strings.TrimSpace(c); c != "" {
|
||||
stated = append(stated, c)
|
||||
}
|
||||
}
|
||||
q.runnerStated = stated
|
||||
}
|
||||
|
||||
@@ -47,27 +47,50 @@ func (q *Session) Asker() Asker {
|
||||
}
|
||||
|
||||
func (l *natsLink) Ping(ctx context.Context, service, id string) error {
|
||||
_, err := l.stated(ctx, service, id)
|
||||
return err
|
||||
}
|
||||
|
||||
// ToolRunnerService is the services-protocol name the tool runner answers discovery under, and CapabilitiesKey the metadata
|
||||
// key of its PING it states the capabilities of the calls it answers under, comma-separated (mesh-tools announce).
|
||||
const (
|
||||
ToolRunnerService = "node-tools"
|
||||
CapabilitiesKey = "capabilities"
|
||||
)
|
||||
|
||||
// Stater is a link that can ask what a service's instance states of itself in its PING: its metadata.
|
||||
type Stater interface {
|
||||
Stated(ctx context.Context, service, id string) (map[string]string, error)
|
||||
}
|
||||
|
||||
// Stated is what the tool runner of this machine states in its PING, asked of its own instance as a witness asks it.
|
||||
func (l *natsLink) Stated(ctx context.Context, service, id string) (map[string]string, error) {
|
||||
return l.stated(ctx, service, id)
|
||||
}
|
||||
|
||||
func (l *natsLink) stated(ctx context.Context, service, id string) (map[string]string, error) {
|
||||
subject := PingSubject(service, id)
|
||||
before := l.conn.LastError()
|
||||
msg, err := l.conn.RequestWithContext(ctx, subject, nil)
|
||||
switch {
|
||||
case errors.Is(err, nats.ErrNoResponders):
|
||||
// The bus delivered the question and nothing subscribes to it: the runtime is not on the bus.
|
||||
return fmt.Errorf("%w: no %s on this machine is on the bus", ErrNoAnswer, service)
|
||||
return nil, fmt.Errorf("%w: no %s on this machine is on the bus", ErrNoAnswer, service)
|
||||
case err != nil:
|
||||
if refused := l.refusal(subject, before); refused != "" {
|
||||
return fmt.Errorf("%w: asking %s: %v%s", ErrCannotAsk, subject, err, refused)
|
||||
return nil, fmt.Errorf("%w: asking %s: %v%s", ErrCannotAsk, subject, err, refused)
|
||||
}
|
||||
return fmt.Errorf("%w: %s did not answer: %v", ErrNoAnswer, subject, err)
|
||||
return nil, fmt.Errorf("%w: %s did not answer: %v", ErrNoAnswer, subject, err)
|
||||
}
|
||||
var ping struct {
|
||||
Name string `json:"name"`
|
||||
ID string `json:"id"`
|
||||
Name string `json:"name"`
|
||||
ID string `json:"id"`
|
||||
Metadata map[string]string `json:"metadata"`
|
||||
}
|
||||
if err := json.Unmarshal(msg.Data, &ping); err != nil || ping.Name != service || ping.ID != id {
|
||||
return fmt.Errorf("%w: what answered %s is not %s %s", ErrNoAnswer, subject, service, id)
|
||||
return nil, fmt.Errorf("%w: what answered %s is not %s %s", ErrNoAnswer, subject, service, id)
|
||||
}
|
||||
return nil
|
||||
return ping.Metadata, nil
|
||||
}
|
||||
|
||||
// refusal is the bus's word on this connection when it refused a publish to subject while this question
|
||||
|
||||
Reference in New Issue
Block a user