diff --git a/internal/compose/resolution/resolve_test.go b/internal/compose/resolution/resolve_test.go index d6f0d4b..7eb06e4 100644 --- a/internal/compose/resolution/resolve_test.go +++ b/internal/compose/resolution/resolve_test.go @@ -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) { diff --git a/internal/link/have.go b/internal/link/have.go index eb13fb0..7f1a2ed 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -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 { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index 2fc4c41..f0825b2 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -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" { diff --git a/internal/link/record.go b/internal/link/record.go index df204af..740184f 100644 --- a/internal/link/record.go +++ b/internal/link/record.go @@ -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 diff --git a/internal/link/session.go b/internal/link/session.go index 3bcd467..7dd16af 100644 --- a/internal/link/session.go +++ b/internal/link/session.go @@ -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 +} diff --git a/internal/link/witnessing.go b/internal/link/witnessing.go index ed364d9..14cd1ec 100644 --- a/internal/link/witnessing.go +++ b/internal/link/witnessing.go @@ -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