mesh-grants: ask the controller's root-free verb and wake on its condition events, so a verified sender is composed (issue 1945)
The part composed every membership with no root-free input, so telegram's verified-sender was dropped and the membership held as a shadow difference. It now asks mesh-controller.root-free for the machines a composition needs, per pass, failing closed and said once, and wakes on root-not-free condition-raised and condition-cleared.
This commit is contained in:
@@ -125,6 +125,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
})
|
||||
}
|
||||
f.Hear = func(handle func(string, []byte)) error {
|
||||
for _, event := range grants.Consumes {
|
||||
event := event
|
||||
if _, err := nc.Subscribe(grants.EventSubject(event), func(m *nats.Msg) { handle(event, m.Data) }); err != nil {
|
||||
return fmt.Errorf("hearing %s: %w", event, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
logger.Printf("started on the account %s: publishing memberships and creating the declared buckets, logs and streams the bus lacks; the user list, grants and any change to a held stream in shadow", account.User)
|
||||
f.Run(ctx)
|
||||
return nil
|
||||
|
||||
@@ -16,6 +16,7 @@ import (
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
@@ -48,6 +49,8 @@ type fakeBus struct {
|
||||
// creates are the streams the part asked to create, refused when refuseCreate.
|
||||
creates []string
|
||||
refuseCreate bool
|
||||
// rootFree answers the controller's root-free verb: nil is no answerer at all (no responders).
|
||||
rootFree func(machines []string) (any, error)
|
||||
}
|
||||
|
||||
func (b *fakeBus) LastMembership(node, module string) ([]byte, uint64, bool, error) {
|
||||
@@ -166,6 +169,15 @@ func (b *fakeBus) Call(key string, body any) (json.RawMessage, error) {
|
||||
return json.Marshal(map[string]any{"users": out})
|
||||
case grants.CallStreams:
|
||||
return json.Marshal(map[string]any{"detail": b.streams})
|
||||
case grants.CallRootFree:
|
||||
if b.rootFree == nil {
|
||||
return nil, errors.New("nats: no responders available for request")
|
||||
}
|
||||
out, err := b.rootFree(args["machines"].([]string))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(out)
|
||||
}
|
||||
return nil, fmt.Errorf("no tool %s", key)
|
||||
}
|
||||
@@ -300,6 +312,7 @@ func TestGrants(t *testing.T) {
|
||||
}{
|
||||
{"reads", m.Reads, grants.Reads},
|
||||
{"invokes", m.Invokes, grants.Invokes},
|
||||
{"consumes", m.Consumes, grants.Consumes},
|
||||
{"emits", m.Emits, grants.Emits},
|
||||
{"tools", m.Tools, grants.Tools},
|
||||
} {
|
||||
@@ -414,6 +427,94 @@ func TestGrants(t *testing.T) {
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("root-free: the controller's word fills the composition, failing closed (issue 1945)", func(t *testing.T) {
|
||||
const router = `{"module":"messenger","version":"1","runs-as":"messenger","resources":[{"id":"account","type":"user","name":"messenger","shell":"/usr/bin/nologin","home":"/var/lib/messenger"}],"own-secrets":{"broker":"/var/lib/m/broker"},"secrets-owner":"messenger",` +
|
||||
`"seats":[{"name":"operator-channel","scope":"mesh","serves":["open"]},{"name":"channel","scope":"mesh","kinded":true,"accepts":["show"]}],` +
|
||||
`"claims":[{"name":"operator-channel","scope":"mesh","serves":["open"]}]}`
|
||||
const telegram = `{"module":"telegram","version":"1","runs-as":"telegram","resources":[{"id":"account","type":"user","name":"telegram","shell":"/usr/bin/nologin","home":"/var/lib/telegram"}],"own-secrets":{"broker":"/var/lib/t/broker"},"secrets-owner":"telegram",` +
|
||||
`"claims":[{"name":"channel","scope":"mesh","kind":"telegram","capabilities":["deliver","verified-sender"]}]}`
|
||||
answer := func(free ...string) func([]string) (any, error) {
|
||||
return func(machines []string) (any, error) {
|
||||
var vs []map[string]any
|
||||
for _, m := range machines {
|
||||
ok := slices.Contains(free, m)
|
||||
why := "judged free"
|
||||
if !ok {
|
||||
why = "its node-engine judged nothing"
|
||||
}
|
||||
vs = append(vs, map[string]any{"machine": m, "free": ok, "why": why})
|
||||
}
|
||||
return map[string]any{"machines": vs}, nil
|
||||
}
|
||||
}
|
||||
for _, row := range []struct {
|
||||
name string
|
||||
routerOn []string
|
||||
telegramOn string
|
||||
ask func([]string) (any, error)
|
||||
keeps bool
|
||||
note string
|
||||
}{
|
||||
{"router and telegram on one free machine", []string{"a"}, "a", answer("a"), true, ""},
|
||||
{"router and telegram on two free machines", []string{"a"}, "b", answer("a", "b"), true, ""},
|
||||
{"the holder's machine is not free", []string{"a"}, "b", answer("a"), false, "b: not free, its node-engine judged nothing"},
|
||||
{"the router's machine is not free", []string{"a"}, "b", answer("b"), false, "a: not free"},
|
||||
{"an error is not free", []string{"a"}, "a", func([]string) (any, error) { return nil, errors.New("boom") }, false, "boom"},
|
||||
{"no answerer is not free", []string{"a"}, "a", nil, false, "no responders"},
|
||||
{"a machine left out of the answer is not free", []string{"a"}, "a",
|
||||
func([]string) (any, error) { return map[string]any{"machines": []any{}}, nil }, false, "named no verdict"},
|
||||
{"the router on two machines frees nothing", []string{"a", "b"}, "a", answer("a", "b"), false, ""},
|
||||
} {
|
||||
t.Run(row.name, func(t *testing.T) {
|
||||
b := fixture()
|
||||
b.manifestOf("messenger", router)
|
||||
b.manifestOf("telegram", telegram)
|
||||
for _, node := range row.routerOn {
|
||||
b.assign(node, "messenger")
|
||||
}
|
||||
b.assign(row.telegramOn, "telegram")
|
||||
b.rootFree = row.ask
|
||||
res := partOn(b).Compose()
|
||||
recs, _ := grants.Records(grants.ReadInputs(b, &grants.Manifests{}))
|
||||
grants.WithRootFree(&recs, b)
|
||||
kept := false
|
||||
for _, k := range buscompose.PlacementsOf(recs, recs.Interchangeable).Kinds {
|
||||
kept = kept || slices.Contains(k.Capabilities, "verified-sender")
|
||||
}
|
||||
if kept != row.keeps {
|
||||
t.Errorf("verified-sender kept = %v, want %v (said %v)", kept, row.keeps, res.RootNotFree)
|
||||
}
|
||||
if !strings.Contains(strings.Join(res.RootNotFree, " | "), row.note) || (row.note == "" && row.keeps && len(res.RootNotFree) != 0) {
|
||||
t.Errorf("the pass says %v, want it to name %q", res.RootNotFree, row.note)
|
||||
}
|
||||
})
|
||||
}
|
||||
// A root-not-free condition raised or cleared wakes a pass; another kind does not (never a timer).
|
||||
wb := fixture()
|
||||
var handle func(string, []byte)
|
||||
wf := &grants.Follower{Part: partOn(wb), Watch: &fakeWatcher{opened: make(chan string, 16)}, Logf: func(string, ...any) {},
|
||||
Hear: func(h func(string, []byte)) error { handle = h; return nil }}
|
||||
wctx, wcancel := context.WithCancel(context.Background())
|
||||
wdone := make(chan struct{})
|
||||
go func() { wf.Run(wctx); close(wdone) }()
|
||||
waitFor(t, func() bool { return wb.count(grants.CallUsers) > 0 && handle != nil })
|
||||
passes := func() int { return wb.count(grants.CallUsers) }
|
||||
before := passes()
|
||||
handle(grants.EventConditionRaised, []byte(`{"kind":"disk-full","key":"machine.a.disk-full"}`))
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
if passes() != before {
|
||||
t.Errorf("a condition of another kind woke a pass")
|
||||
}
|
||||
handle(grants.EventConditionCleared, []byte(`{"kind":"root-not-free","key":"machine.a.root-not-free"}`))
|
||||
waitFor(t, func() bool { return passes() > before })
|
||||
wcancel()
|
||||
<-wdone
|
||||
if !grants.IsRootNotFree([]byte(`{"kind":"root-not-free"}`)) || !grants.IsRootNotFree([]byte(`{"key":"machine.novox.root-not-free"}`)) ||
|
||||
grants.IsRootNotFree([]byte(`{"kind":"disk-full","key":"machine.novox.disk-full"}`)) {
|
||||
t.Errorf("only a root-not-free condition wakes a pass")
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("a predefined-key fallback is read with no list; every refused manifest listed", func(t *testing.T) {
|
||||
// mod-p's value falls back to a predefined key, as claude-code's managed_settings does, and no list is on the bus
|
||||
// (issue 1945); mod-z's manifest carries a key mesh-sdk's reader refuses, and mod-z is assigned nowhere.
|
||||
|
||||
@@ -28,14 +28,20 @@ type Follower struct {
|
||||
Logf func(string, ...any)
|
||||
// Asked watches the part's own ask to compose now (a tool's compose-now), waking at each; nil watches none.
|
||||
Asked func(wake func()) error
|
||||
// Hear hears the controller seat's condition events (Consumes) and hands each, by event name, with its body, to
|
||||
// handle; nil hears none. A condition of the kind root-not-free raised or cleared wakes a pass: a machine's
|
||||
// root-free word changed.
|
||||
Hear func(handle func(event string, body []byte)) error
|
||||
|
||||
wake chan struct{}
|
||||
once sync.Once
|
||||
mu sync.Mutex
|
||||
asked bool
|
||||
watched map[string]bool
|
||||
refused map[string]string
|
||||
have map[string]string
|
||||
wake chan struct{}
|
||||
once sync.Once
|
||||
mu sync.Mutex
|
||||
asked bool
|
||||
heard_ bool
|
||||
rootSaid string
|
||||
watched map[string]bool
|
||||
refused map[string]string
|
||||
have map[string]string
|
||||
// held and notAsserted are what the last pass held back and could not create, said again only when they change.
|
||||
held string
|
||||
notAsserted string
|
||||
@@ -116,6 +122,18 @@ func (f *Follower) open() {
|
||||
f.asked = true
|
||||
}
|
||||
}
|
||||
if f.Hear != nil && !f.heard_ {
|
||||
err := f.Hear(func(event string, body []byte) {
|
||||
if IsRootNotFree(body) {
|
||||
f.Wake()
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
f.logf("the controller's condition events cannot be heard yet (%v): tried again at the next change", err)
|
||||
} else {
|
||||
f.heard_ = true
|
||||
}
|
||||
}
|
||||
for _, state := range Followed {
|
||||
f.mu.Lock()
|
||||
open := f.watched[state]
|
||||
@@ -175,6 +193,15 @@ func (f *Follower) said(res Result) {
|
||||
}
|
||||
}
|
||||
}
|
||||
if said := strings.Join(res.RootNotFree, "\n"); said != f.rootSaid {
|
||||
f.rootSaid = said
|
||||
if len(res.RootNotFree) == 0 {
|
||||
f.logf("every machine a composition needs a root-free word for is free")
|
||||
}
|
||||
for _, n := range res.RootNotFree {
|
||||
f.logf("root-free: %s (a verified sender is composed as from no root-free machine there)", n)
|
||||
}
|
||||
}
|
||||
for _, a := range res.Asserted {
|
||||
f.logf("asserted %s", a)
|
||||
}
|
||||
|
||||
@@ -46,7 +46,7 @@ const (
|
||||
)
|
||||
|
||||
// Invokes is every tool the part calls.
|
||||
var Invokes = []string{CallUsers, CallUserCan, CallStreams}
|
||||
var Invokes = []string{CallUsers, CallUserCan, CallStreams, CallRootFree}
|
||||
|
||||
// Reader reads another part's state, named `<module>.<state>`. A refusal (a read not granted yet, a bucket not there) is
|
||||
// an error the part names as a gap and composes past: it never stops for one.
|
||||
|
||||
@@ -22,7 +22,7 @@ import (
|
||||
// MembershipGaps are what a membership carries that the part cannot compose from the bus.
|
||||
var MembershipGaps = []Gap{
|
||||
{Input: "membership: private network and received contributions", Why: "kept in each node's composition, which only that node reads: carried over from the membership last published, none for a first one"},
|
||||
{Input: "membership: root-free machines", Why: "judged by the controller on the bus at its composition, not kept: a verified sender is composed as from no root-free machine"},
|
||||
{Input: "membership: root-free machines", Why: "asked of the controller's root-free verb at each pass and kept for that pass alone; a machine it does not call free (an error included) is composed as not free, and the pass says why"},
|
||||
}
|
||||
|
||||
// Publisher publishes a membership on the assignments stream (membership.Subject), last per subject, and reads the one
|
||||
|
||||
@@ -100,7 +100,10 @@ type Result struct {
|
||||
NotAsserted []string `json:"not_asserted,omitempty"`
|
||||
// Refused is every manifest the indexer holds that this pass could not read, `manifest of <module>: <why>`: its
|
||||
// module counts as placed nowhere (issue 1945).
|
||||
Refused []string `json:"refused,omitempty"`
|
||||
Refused []string `json:"refused,omitempty"`
|
||||
// RootNotFree is each machine the composition needed a root-free word for and did not get one, `<machine>: not
|
||||
// free, <why>`: a verified sender is composed as from no root-free machine there (issue 1945).
|
||||
RootNotFree []string `json:"root_not_free,omitempty"`
|
||||
Errors []string `json:"errors,omitempty"`
|
||||
PublishedOrAsserted string `json:"published_or_asserted"`
|
||||
}
|
||||
@@ -172,6 +175,7 @@ func (p *Part) Compose() Result {
|
||||
res.Refused = append(res.Refused, g.Input+": "+g.Why)
|
||||
}
|
||||
recs, left := Records(in)
|
||||
res.RootNotFree = WithRootFree(&recs, p.Call)
|
||||
st := UserListState{V: 1, Mode: Shadow, LeftOut: left, Loaded: in.Loaded, BusNode: in.BusNode,
|
||||
Gaps: append(append([]Gap(nil), in.Gaps...), MembershipGaps...), At: p.now()}
|
||||
if st.LeftOut == nil {
|
||||
|
||||
@@ -0,0 +1,122 @@
|
||||
package grants
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/buscompose"
|
||||
)
|
||||
|
||||
// CallRootFree is the controller seat's verb that judges machines root-free (novox/hq ADR 0259 §8): the one judgement
|
||||
// the composer needs and the part cannot make from the bus (issue 1945). Only the controller answers it, on purpose.
|
||||
const CallRootFree = "seat:mesh-controller.root-free"
|
||||
|
||||
// The controller seat's condition events the part hears (never a timer): a machine's root-free word changes with the
|
||||
// raising and the clearing of its `root-not-free` condition.
|
||||
const (
|
||||
EventConditionRaised = "mesh-controller.condition-raised"
|
||||
EventConditionCleared = "mesh-controller.condition-cleared"
|
||||
// KindRootNotFree is the condition kind whose events wake a pass.
|
||||
KindRootNotFree = "root-not-free"
|
||||
)
|
||||
|
||||
// Consumes is every event the part hears, as its manifest's consumes name them.
|
||||
var Consumes = []string{EventConditionRaised, EventConditionCleared}
|
||||
|
||||
// EventSubject is the bus subject of a controller seat event named `mesh-controller.<event>`.
|
||||
func EventSubject(event string) string {
|
||||
return "mesh.seat.mesh-controller.event." + strings.TrimPrefix(event, "mesh-controller.")
|
||||
}
|
||||
|
||||
// IsRootNotFree says a condition event's body names the `root-not-free` kind (its `kind`, or the last part of its
|
||||
// `key`, `<scope>.<id>.<kind>`). An unreadable body wakes a pass too: a pass is cheap, a missed change is not.
|
||||
func IsRootNotFree(body []byte) bool {
|
||||
var c struct {
|
||||
Kind string `json:"kind"`
|
||||
Key string `json:"key"`
|
||||
}
|
||||
if json.Unmarshal(body, &c) != nil {
|
||||
return true
|
||||
}
|
||||
return c.Kind == KindRootNotFree || (c.Kind == "" && strings.HasSuffix(c.Key, "."+KindRootNotFree)) || (c.Kind == "" && c.Key == "")
|
||||
}
|
||||
|
||||
// RootFreeNodes are the machines a composition needs a root-free word for: where the router's seat is held and where a
|
||||
// kinded seat's holder declares `verified-sender`. None when no holder declares it. Sorted. Pure.
|
||||
func RootFreeNodes(r buscompose.Records) []string {
|
||||
var nodes []string
|
||||
need := false
|
||||
for node, declared := range r.Assigned {
|
||||
for _, d := range declared {
|
||||
for _, s := range d.Holds {
|
||||
if s.Name == "operator-channel" && !slices.Contains(nodes, node) {
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
if s.Kinded && s.Kind != "" && slices.Contains(s.Capabilities, "verified-sender") {
|
||||
need = true
|
||||
if !slices.Contains(nodes, node) {
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if !need {
|
||||
return nil
|
||||
}
|
||||
sort.Strings(nodes)
|
||||
return nodes
|
||||
}
|
||||
|
||||
// verdict is one machine's answer of the controller's root-free verb.
|
||||
type verdict struct {
|
||||
Machine string `json:"machine"`
|
||||
Free bool `json:"free"`
|
||||
Why string `json:"why"`
|
||||
}
|
||||
|
||||
// WithRootFree fills r.RootFree for the machines the composition needs by asking the controller once, and answers a
|
||||
// line for each machine that is not free, `<machine>: <why>`. It fails closed: an error, no answer or a machine
|
||||
// left out is not free, and says so. Pure but for the one call; the answer is kept for this pass alone.
|
||||
func WithRootFree(r *buscompose.Records, call Caller) []string {
|
||||
nodes := RootFreeNodes(*r)
|
||||
if len(nodes) == 0 {
|
||||
return nil
|
||||
}
|
||||
free := map[string]verdict{}
|
||||
var failure string
|
||||
if call == nil {
|
||||
failure = "no way to ask the controller here"
|
||||
} else if raw, err := call.Call(CallRootFree, map[string]any{"machines": nodes}); err != nil {
|
||||
failure = "the controller's root-free word could not be asked: " + err.Error()
|
||||
} else {
|
||||
var body struct {
|
||||
Machines []verdict `json:"machines"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &body); err != nil {
|
||||
failure = "the controller's root-free answer could not be read: " + err.Error()
|
||||
}
|
||||
for _, v := range body.Machines {
|
||||
free[v.Machine] = v
|
||||
}
|
||||
}
|
||||
r.RootFree = map[string]bool{}
|
||||
var notes []string
|
||||
for _, n := range nodes {
|
||||
v, answered := free[n]
|
||||
switch {
|
||||
case answered && v.Free:
|
||||
r.RootFree[n] = true
|
||||
case failure != "":
|
||||
notes = append(notes, fmt.Sprintf("%s: not free, %s", n, failure))
|
||||
case !answered:
|
||||
notes = append(notes, fmt.Sprintf("%s: not free, the controller named no verdict for it", n))
|
||||
default:
|
||||
notes = append(notes, fmt.Sprintf("%s: not free, %s", n, v.Why))
|
||||
}
|
||||
}
|
||||
return notes
|
||||
}
|
||||
@@ -70,8 +70,9 @@ func (p *Part) MembershipOf(node, module string) (any, error) {
|
||||
}
|
||||
in := ReadInputs(p.Read, &p.cache)
|
||||
recs, left := Records(in)
|
||||
notFree := WithRootFree(&recs, p.Call)
|
||||
m, ok := Memberships(recs)[node+"."+module]
|
||||
out := map[string]any{"node": node, "module": module, "gaps": MembershipGaps}
|
||||
out := map[string]any{"node": node, "module": module, "gaps": MembershipGaps, "root_free": recs.RootFree, "root_not_free": notFree}
|
||||
if ok {
|
||||
out["composed"] = m
|
||||
} else {
|
||||
|
||||
@@ -13,10 +13,15 @@
|
||||
"node-engine.have",
|
||||
"mesh-vault.bus-accounts"
|
||||
],
|
||||
"consumes": [
|
||||
"mesh-controller.condition-raised",
|
||||
"mesh-controller.condition-cleared"
|
||||
],
|
||||
"invokes": [
|
||||
"nats.nats_streams",
|
||||
"nats.nats_user_can",
|
||||
"nats.nats_users"
|
||||
"nats.nats_users",
|
||||
"seat:mesh-controller.root-free"
|
||||
],
|
||||
"emits": [
|
||||
"asserted",
|
||||
|
||||
Reference in New Issue
Block a user