W8.7 rest, W8.6 (node place): the grants part keeps what each node-engine runs (running, grants_running) and reads seat holders from its digest; a machine's place in an event's subject is the definition's <node>

The part binds its own durable on the node-engine's kept stream (HearKept, tried at every wake until the pass has made it),
the seat's holders are read from the manifest of the digest each node says it runs, and a node-engine's account is granted its
own node at the part its definition names (running.<node>.*) and its durable on the kept stream by name.

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

Builds: design 47 G §199, §249, §309
This commit is contained in:
2026-11-11 14:40:26 +01:00
parent 28ebd45863
commit 4a61381ff1
18 changed files with 447 additions and 20 deletions
@@ -367,6 +367,45 @@ func TestTheGrantsPartWhole(t *testing.T) {
info, err := cons.Info(ctx)
return err == nil && slices.Contains(info.Config.FilterSubjects, "mesh.seat.mesh-broker.event.created")
})
// Design 47, ADR 0660 §2: the seat node-engine's running is kept on a stream of its own, sourced from EVENTS (the bus
// takes one stream a subject), last message per subject. The part makes it, says nothing of it on EVENTS' consumers, and
// keeps what its own durable on it hears in its state running, which holds the last said even for a running said
// before the durable was made.
until("the kept stream of running, sourced from EVENTS, made by the part", func() bool {
st, err := js.Stream(ctx, nodeengine.StreamRunning)
if err != nil {
return false
}
cfg := st.CachedInfo().Config
return len(cfg.Subjects) == 0 && len(cfg.Sources) == 1 && cfg.Sources[0].Name == "EVENTS" && cfg.MaxMsgsPerSubject == 1
})
say := func(node, module, digest string) {
body, _ := json.Marshal(nodeengine.RunningEvent{V: 1, At: time.Now(), Fingerprint: "fp-" + digest, Manifest: digest, Module: module, Node: node})
if _, err := js.Publish(ctx, nodeengine.SubjectRunning(node, module), body); err != nil {
t.Fatal(err)
}
}
say("a", "mod-c", "sha256:one")
say("a", "mod-c", "sha256:two")
until("the state running holding the last running of a.mod-c, heard on the part's durable of the kept stream", func() bool {
e, err := kvs[grants.Module+"."+grants.StateRunning].Get(ctx, "a.mod-c")
if err != nil {
return false
}
var r nodeengine.RunningEvent
return json.Unmarshal(e.Value(), &r) == nil && r.Manifest == "sha256:two"
})
if info, err := js.Stream(ctx, "EVENTS"); err == nil {
if cons, err := js.Consumer(ctx, "EVENTS", "a_"+grants.Module); err == nil {
ci, _ := cons.Info(ctx)
for _, f := range ci.Config.FilterSubjects {
if f == nodeengine.SubjectRunningAll {
t.Errorf("the part's durable on EVENTS still filters %s: running is heard on its kept stream alone", f)
}
}
}
_ = info
}
bindingOf := func() []membership.BusBinding {
e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c")
if err != nil {
@@ -28,12 +28,14 @@ import (
"os"
"os/signal"
"strings"
"sync"
"syscall"
"time"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/announce"
@@ -136,6 +138,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Objects: sealedObjects{ctx: ctx, js: js}, Purging: ownPurging{ctx: ctx, js: js, module: account.Module}, Logf: logger.Printf,
Bus: grants.BusAddress{URL: account.URL, Fingerprint: account.Fingerprint}}
f.Watch = fv
f.HearKept = func() error { return hearRunning(ctx, account, js, f, logger.Printf) }
// The first view of the bindings (design 25 bindings-first-view): the seat mesh-broker's list, asked at each start and by
// hand, with the seat's events held until its answer is applied or its bound passes.
f.Part.Bindings = &grants.BindingsView{Say: sayCond, Changed: f.Wake, Logf: logger.Printf,
@@ -255,3 +258,37 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
f.Run(ctx)
return nil
}
// hearing is the stream the part's running consumer is bound to, once.
var hearingRunning struct {
sync.Mutex
cc jetstream.ConsumeContext
}
// hearRunning binds the part's durable on the node-engine's kept stream running and keeps each event in the state running,
// then wakes a pass. It binds once; a call while it is bound does nothing, and an error (the stream or durable not made yet)
// is tried again at the next pass.
func hearRunning(ctx context.Context, account process.Account, js jetstream.JetStream, f *grants.Follower, logf func(string, ...any)) error {
hearingRunning.Lock()
defer hearingRunning.Unlock()
if hearingRunning.cc != nil {
return nil
}
cc, err := account.HearSeatEvents(ctx, js, nodeengine.StreamRunning, func(m jetstream.Msg) {
changed, err := f.Part.HearRunning(m.Subject(), m.Data())
if err != nil {
logf("running: %v, dropped", err)
_ = m.Term()
return
}
_ = m.Ack()
if changed {
f.Wake()
}
}, logf)
if err != nil {
return err
}
hearingRunning.cc = cc
return nil
}
@@ -248,6 +248,12 @@ func toolsOf(p *grants.Part) []stdio.Tool {
Run: func(a map[string]any) (any, error) {
return p.AskFirstViewAgain(stdio.CallerOf(a), grants.Answer(a, "state"), grants.Answer(a, "why"), recordHandAct)
}},
{Name: grants.ToolRunning, Description: "What each node-engine said it runs (novox/hq design 47, ADR 0660 §2): the last event " +
"running of each node and module, as the grants part keeps it in its state running (module, node, the build's " +
"fingerprint, the digest of that build's manifest, when it was seen running healthy). A seat's holder is read from " +
"the manifest of that digest; a node that said nothing is read from the trunk's manifest. One key, `<node>.<module>`, or all. Read-only.",
Input: schema(map[string]string{"key": "one `<node>.<module>`; none for all"}),
Run: func(a map[string]any) (any, error) { return p.RunningOf(grants.Answer(a, "key")) }},
}
}
+48
View File
@@ -1343,6 +1343,11 @@ func TestGrants(t *testing.T) {
{"the holder's running build still claims it", map[string]catalogue.Manifest{"a.mesh-catalog": claimer("mesh-catalog")}, nil, []string{"mesh-catalog"}, ""},
{"a second claimant", nil, []string{"other"}, nil, "seat mesh-catalog --to <node>/<module>"},
}
// What the want names is no running build: the seat's holder is read from the digest the node-engine said it runs.
wantedOnly := rows[1]
wantedOnly.name = "a build only wanted, not said running by its node-engine, is not read"
wantedOnly.want, wantedOnly.said = []string{"mesh-delivery-indexer"}, "composed as held by mesh-delivery-indexer"
rows = append(rows, wantedOnly)
for _, row := range rows {
t.Run(row.name, func(t *testing.T) {
in := grants.Inputs{Nodes: []string{"a", "b"},
@@ -1353,6 +1358,9 @@ func TestGrants(t *testing.T) {
"a": {Node: "a", Modules: []meshmoduleassigner.PlacedModule{{Module: "mesh-catalog"}},
Holds: []meshmoduleassigner.Holding{{Claim: "mesh-catalog", Scope: "mesh", Module: "mesh-catalog"}}},
"b": {Node: "b", Modules: []meshmoduleassigner.PlacedModule{{Module: "mesh-delivery-indexer"}}}}}
if !strings.Contains(row.name, "only wanted") {
in.Ran = row.running
}
maps.Copy(in.Running, row.running)
for _, c := range row.extra {
in.Manifests[c] = claimer(c)
@@ -4760,3 +4768,43 @@ func channelTakes(t *testing.T, say func(oc.Condition) error) func(oc.Condition)
return say(c)
}
}
// Design 47, ADR 0660 §2: what the part hears of the node-engines' running is kept, a key a node and module, only when it
// differs; a body that does not match its subject, or names no manifest, is refused; grants_running reads it back.
func TestTheGrantsPartKeepsWhatEachNodeRuns(t *testing.T) {
body := func(node, module, fp, digest string) []byte {
raw, _ := json.Marshal(nodeengine.RunningEvent{V: 1, At: time.Now(), Fingerprint: fp, Manifest: digest, Module: module, Node: node})
return raw
}
bus := newFakeBus()
p := &grants.Part{Store: ownStore{bus}}
subject := nodeengine.SubjectRunning("ace", "mod-c")
for _, row := range []struct {
name, subject string
data []byte
changed bool
refused string
}{
{"a first running is kept", subject, body("ace", "mod-c", "fp1", "sha256:a"), true, ""},
{"the same build said again keeps nothing", subject, body("ace", "mod-c", "fp1", "sha256:a"), false, ""},
{"a new build is kept", subject, body("ace", "mod-c", "fp2", "sha256:b"), true, ""},
{"a body of another node than its subject is refused", subject, body("g14", "mod-c", "fp2", "sha256:b"), false, "does not carry"},
{"a body with no manifest is refused", subject, body("ace", "mod-c", "fp2", ""), false, "no node, module or manifest"},
{"a body that is no running is refused", subject, []byte("x"), false, "not a running body"},
} {
t.Run(row.name, func(t *testing.T) {
changed, err := p.HearRunning(row.subject, row.data)
if changed != row.changed || (err == nil) != (row.refused == "") || err != nil && !strings.Contains(err.Error(), row.refused) {
t.Fatalf("HearRunning = %v, %v; want %v, %q", changed, err, row.changed, row.refused)
}
})
}
got, err := p.RunningOf("")
if err != nil {
t.Fatal(err)
}
said := got.(grants.RunningSaid).Running
if len(said) != 1 || said[0].Manifest != "sha256:b" {
t.Errorf("grants_running = %+v, want the last build of ace.mod-c", said)
}
}
@@ -1294,3 +1294,46 @@ func TestAOneFilterConsumerIsCreatedOnThePlainSubject(t *testing.T) {
t.Fatalf("a consumer with no filter is created with filters %q %v", none.FilterSubject, none.FilterSubjects)
}
}
// Design 47: a node-scoped seat's event names its machine at the part its definition calls `<node>`: the node-engine's account
// is granted its own node there and any module, a watcher any node and any module, and the engine's durable on a kept stream
// is asked about, pulled and acknowledged by name. Before, the node was always the last token (`running.*.*.*`).
func TestAMachineEventNamesItsNodeWhereTheDefinitionSaysSo(t *testing.T) {
engine := Seat{Name: EngineSeat, Scope: "node", HeldByTheEngine: true,
EventSubjects: map[string]string{"running.*.*": "running.<node>.<module>", "composed.*": "composed.<node>", "owed.*": "owed.<module>"}}
for _, row := range []struct {
name, event, want string
ok bool
}{
{"the node is the first part", "running.*.*", "running.<node>.*", true},
{"the node is the only part", "composed.*", "composed.<node>", true},
{"a part that is no node's says no machine", "owed.*", "", false},
{"an event with no subject takes the node last", "done", "done.<node>", true},
} {
t.Run(row.name, func(t *testing.T) {
got, ok := machineEvent(engine, row.event)
if got != row.want || ok != row.ok {
t.Fatalf("machineEvent = %q, %v; want %q, %v", got, ok, row.want, row.ok)
}
if ok && anyMachine(got) == onMachine(got, "ace") {
t.Errorf("a watcher and the holder are granted the same: %q", got)
}
})
}
kept := KeptEvent{Seat: EngineSeat, Event: "running", Stream: "SEAT_node-engine_running", Filter: "mesh.seat.node-engine.event.running.*.*"}
pub, sub := engineGrants(Principal{Kind: KindNode, Node: "ace", EngineEvents: engine.EventSubjects, EngineKept: []KeptEvent{kept}})
for _, want := range []string{"mesh.seat.node-engine.event.running.ace.*", "$JS.API.CONSUMER.CREATE.SEAT_node-engine_running.ace",
"$JS.API.CONSUMER.INFO.SEAT_node-engine_running.ace", "$JS.API.CONSUMER.MSG.NEXT.SEAT_node-engine_running.ace", "$JS.ACK.SEAT_node-engine_running.ace.>"} {
if !slices.Contains(pub, want) {
t.Errorf("the engine of ace is not granted %s", want)
}
}
for _, bad := range []string{"mesh.seat.node-engine.event.running.*.*", "mesh.seat.node-engine.event.running.g14.*"} {
if slices.Contains(pub, bad) {
t.Errorf("the engine of ace is granted %s", bad)
}
}
if !slices.Contains(sub, kept.Filter) {
t.Errorf("the engine does not hear every node's running: %v", sub)
}
}
@@ -102,7 +102,7 @@ func keeperConsumer(p Principal) (Consumer, bool) {
}
for _, e := range w.Emits {
if pattern, ok := machineEvent(w, e); ok {
e = pattern
e = anyMachine(pattern)
}
filters = append(filters, seatSubject(w, "event", e))
}
@@ -4,6 +4,8 @@ import (
nodeengineseat "git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
"git.novox.be/novox/mesh-delivery/modules/mesh-grants/busgrants"
"maps"
"slices"
)
// The node seat node-engine (novox/hq ADR 0563 §1): every node's engine holds it on its own bus account, and no module
@@ -60,6 +62,29 @@ func engineGrants(p Principal) (pub, sub []string) {
// Any event of the seat about its own node, by the one rule a watcher of it is composed by (machineEvent).
every, _ := machineEvent(Seat{Name: EngineSeat, Scope: "node", HeldByTheEngine: true}, "*")
pub = append(pub, seatSubject(Seat{Name: EngineSeat}, "event", onMachine(every, p.Node)))
// And an event whose subject carries parts says its machine at the part the definition names `<node>` (design 47:
// `running.<node>.<module>`): its own node's alone, any module, never another node's.
seat := Seat{Name: EngineSeat, Scope: "node", HeldByTheEngine: true, EventSubjects: p.EngineEvents}
for _, e := range slices.Sorted(maps.Keys(p.EngineEvents)) {
if pattern, ok := machineEvent(seat, e); ok {
pub = append(pub, seatSubject(seat, "event", onMachine(pattern, p.Node)))
}
}
// And the durable it makes on each of the seat's events kept in a stream of its own (the engine is no module, so no
// Heard): its own, named as its consumer on the shared stream is, asked about, pulled and acknowledged, and the seat's
// events of every node heard there, since a seat's pass is decided from what the other nodes run.
if safeSubject.MatchString(p.Node) {
for _, k := range p.EngineKept {
if !safeSubject.MatchString(k.Stream) {
continue
}
pub = append(pub,
"$JS.API.CONSUMER.CREATE."+k.Stream+"."+p.Node, "$JS.API.CONSUMER.CREATE."+k.Stream+"."+p.Node+".>",
"$JS.API.CONSUMER.INFO."+k.Stream+"."+p.Node, "$JS.API.CONSUMER.MSG.NEXT."+k.Stream+"."+p.Node,
"$JS.ACK."+k.Stream+"."+p.Node+".>")
sub = append(sub, k.Filter)
}
}
sub = append(sub, announcing(EngineSeat)...)
pub = append(pub, discovering()...)
tp, ts := seatTrafficGrants(SeatTrafficOf(occ.EngineAsker(p.Node), nil, p.Uses, nil))
@@ -106,6 +106,10 @@ type Seat struct {
// HeldByTheEngine says every node's engine holds the seat on its own account (catalogue.Seat.HeldByTheEngine, novox/hq
// ADR 0563 §1): each of its events names the engine's machine as its last token (machineEvent).
HeldByTheEngine bool
// EventSubjects are the subjects of the seat's events that carry parts, by the event's local name (`running.*.*`), as the
// definition gives them (`running.<node>.<module>`): where machineEvent finds the machine. Empty where the seat list is
// unread.
EventSubjects map[string]string
}
// A Principal is one user of the bus. Its permissions are derived from what it declares and
@@ -115,6 +119,11 @@ type Principal struct {
Kind Kind
Node string
Module string
// EngineEvents are the node-engine seat's events with parts in their subject, by local name (Seat.EventSubjects), and
// EngineKept those it keeps in a stream of its own, for a KindNode principal: what its engine's account is granted
// (engineGrants).
EngineEvents map[string]string
EngineKept []KeptEvent
// Ledgers are the local names of the ledgers its module asks for in `requires`, for a KindLedger principal alone
// (issue 1966 D5): its whole authority is publishing on those ledgers' histories and snapshots (LedgerPublishGrants).
Ledgers []string
@@ -427,24 +436,49 @@ func NodeLoginsReads(node string) []string {
// holder's machine (novox/hq ADR 0219): `paused.<node>`, the build agent saying whether it takes work.
var perMachineEvents = map[string]bool{"paused.*": true}
// machineEvent is the one rule for where a node-scoped seat's event names its holder's machine as its last token (novox/hq
// ADR 0219, ADR 0618 §4), as a pattern whose last token `*` is that machine, and whether e is such an event: every event
// of a seat the node-engine holds (engineGrants), and each of perMachineEvents of any other node seat. The holder is
// granted the pattern with its own machine in place of `*`, a watcher the pattern, so the two cannot differ.
// nodeToken is the place of a machine in a machineEvent pattern: the holder is granted its own machine's name there, a
// watcher any (anyMachine).
const nodeToken = "<node>"
// machineEvent is the one rule for where a node-scoped seat's event names its holder's machine (novox/hq ADR 0219, ADR 0618
// §4; design 47), as a pattern whose token `<node>` is that machine, and whether e is such an event: every event of a seat
// the node-engine holds (engineGrants) at the part of its subject the definition names `<node>` (Seat.EventSubjects:
// `running.<node>.<module>`), or last when its subject carries none, and each of perMachineEvents of any other node seat.
// The holder is granted the pattern with its own machine in place of the token, a watcher with `*` (anyMachine), so the
// two cannot differ, and a node is granted only subjects carrying its own name.
func machineEvent(s Seat, e string) (string, bool) {
switch {
case s.Scope != "node":
return "", false
case perMachineEvents[e]:
return e, true
return strings.TrimSuffix(e, "*") + nodeToken, true
case s.HeldByTheEngine:
return e + ".*", true
if subject, ok := s.EventSubjects[e]; ok {
toks := strings.Split(subject, ".")
found := false
for i, t := range toks {
switch {
case t == nodeToken:
found = true
case len(t) > 2 && t[0] == '<':
toks[i] = "*"
}
}
if !found {
return "", false
}
return strings.Join(toks, "."), true
}
return e + "." + nodeToken, true
}
return "", false
}
// onMachine is a machineEvent pattern with node in place of its last `*`.
func onMachine(pattern, node string) string { return strings.TrimSuffix(pattern, "*") + node }
// anyMachine is a machineEvent pattern a watcher hears: every machine's.
func anyMachine(pattern string) string { return strings.ReplaceAll(pattern, nodeToken, "*") }
// onMachine is a machineEvent pattern with node in place of the machine's token.
func onMachine(pattern, node string) string { return strings.ReplaceAll(pattern, nodeToken, node) }
// enrolmentPrefix is the space every enrolling node's user and inbox live under, so the one place the
// controller may answer an enrolment is derived from the same constant the user is named from.
@@ -1629,7 +1663,7 @@ func watchedEvents(watches []Seat) []string {
}
for _, e := range w.Emits {
if pattern, ok := machineEvent(w, e); ok {
e = pattern
e = anyMachine(pattern)
}
out = append(out, seatSubject(w, "event", e))
}
@@ -149,6 +149,9 @@ type Records struct {
// Kept is every seat event kept in a stream of its own (kept.go): the grants part makes each stream, so its principal is
// granted the create of each by name.
Kept []KeptEvent
// EngineEvents are the node-engine seat's events that carry parts in their subject, by local name: where a node's own
// name sits in each (machineEvent).
EngineEvents map[string]string
// Nodes is every machine the mesh knows. Each gets a host user.
Nodes []string
// Assigned is the modules on each node, as they declare themselves.
@@ -217,7 +220,7 @@ func Users(r Records) ([]Principal, error) {
}
for _, login := range nodeLoginsOf(r, node) {
out = append(out, Principal{Kind: KindNode, Node: node, Login: login, Checks: checks, Runner: runner, Consumes: engineHears,
Watches: engineWatches, Invokes: engineInvokes,
Watches: engineWatches, Invokes: engineInvokes, EngineEvents: r.EngineEvents, EngineKept: keptOfSeat(r.Kept, EngineSeat),
MovesIntoTheVault: servesMoveIn(r), StatesItsKeyToTheVault: servesVerb(r, VaultNodeKey),
AsksItsOwnSecrets: servesVerb(r, VaultOwn), AsksTheVaultToSign: servesVerb(r, VaultSign), Uses: asksSeatOf(r),
FollowsGrantsUserList: bus, ServedReads: servedOn(served, node), RotatesItsLogin: servesVerb(r, VaultRotateNode)})
@@ -643,3 +646,14 @@ func servedOn(served []string, node string) []string {
}
return out
}
// keptOfSeat is the kept events of one seat.
func keptOfSeat(kept []KeptEvent, seat string) []KeptEvent {
var out []KeptEvent
for _, k := range kept {
if k.Seat == seat {
out = append(out, k)
}
}
return out
}
+13 -1
View File
@@ -36,12 +36,17 @@ type Follower struct {
// 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; and so does an account the vault minted (ADR 0561 §1), whose hash the list carries.
Hear func(handle func(event string, body []byte)) error
// HearKept hears a seat's event kept on a stream of its own (running.go): it is tried at every wake until it binds the
// part's durable there, which the pass makes from the manifest's consumes, so the first pass that makes it is followed by one
// more; nil hears none.
HearKept func() error
wake chan struct{}
once sync.Once
mu sync.Mutex
asked bool
heard_ bool
keptHeld bool
rootSaid string
watched map[string]bool
refused map[string]string
@@ -153,6 +158,13 @@ func (f *Follower) open() {
f.heard_ = true
}
}
if f.HearKept != nil && !f.keptHeld {
if err := f.HearKept(); err != nil {
f.logf("what each node runs cannot be heard yet (%v): tried again at the next change", err)
} else {
f.keptHeld = true
}
}
for _, state := range Followed {
f.mu.Lock()
open := f.watched[state]
@@ -252,7 +264,7 @@ func (f *Follower) Run(ctx context.Context) {
f.said(res)
// A consumer this pass created may be the part's own, which its hearing waits on: one more pass opens it, so
// the first event on it is heard without waiting for an unrelated change.
if f.Hear != nil && !f.heard_ && res.Consumers != nil && len(res.Consumers.Created) > 0 {
if (f.Hear != nil && !f.heard_ || f.HearKept != nil && !f.keptHeld) && res.Consumers != nil && len(res.Consumers.Created) > 0 {
f.Wake()
}
if res.Written {
@@ -155,6 +155,9 @@ type Inputs struct {
// Running is the manifest a machine is wanted to run, by `<node>.<module>`: the one the builder built at the
// fingerprint the assigner's want names. An assignment with none is composed from the trunk's manifest.
Running map[string]catalogue.Manifest
// Ran is the manifest of the digest each node-engine says it runs, by `<node>.<module>` (running.go): what a seat's
// holder is read from. A node that said nothing has none, and its module is read from the trunk's manifest.
Ran map[string]catalogue.Manifest
// Had is the manifest of the build each machine runs now (its have's fingerprint, the builder's build of it), by
// `<node>.<module>`; the want's Running may already be a newer build. A membership is composed from both and the trunk's.
Had map[string]catalogue.Manifest
@@ -180,6 +183,9 @@ type Inputs struct {
// Kept is every seat event the seat list says is kept in a stream of its own (buscompose.KeptEventsOf): the part makes
// each stream and each hearing module's durable on it. Empty while the list is unread.
Kept []buscompose.KeptEvent
// EventSubjects are, by seat, the subjects of its events that carry parts, by the event's local name, as the definitions
// give them: where a node-scoped seat's event names its machine (buscompose.Seat.EventSubjects).
EventSubjects map[string]map[string]string
// SeatListUnread says the seat list has not been read (the gate says so, seatlist.Reader.MayAct with no seat): only the
// bootstrapped seats compose, so the pass keeps what the bus loaded (keeploaded.go).
SeatListUnread bool
@@ -345,6 +351,7 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
in.SeatListUnread = gate != nil && !gate.MayAct()
if d, ok := gate.(SeatDefinitions); ok && !in.SeatListUnread {
in.Kept = buscompose.KeptEventsOf(d.Definitions())
in.EventSubjects = eventSubjectsOf(d.Definitions())
}
nodes := map[string]bool{}
// **The seat's assignments alone places modules** (novox/hq ADR 0549 §5, ADR 0567 §9; issue 774): the controller's
@@ -84,3 +84,33 @@ func createKept(c Creator, done *creations, kept []buscompose.KeptEvent, held ma
}
return made, notMade
}
// eventSubjectsOf is, by seat, each event's subject pattern (`running.<node>.<module>`) of the events whose subject carries
// parts, by the event's local name (`running.*.*`).
func eventSubjectsOf(defs map[string]seatdef.Definition) map[string]map[string]string {
out := map[string]map[string]string{}
for _, d := range defs {
for _, e := range d.Events {
if len(e.Parts()) == 0 {
continue
}
if out[d.Seat] == nil {
out[d.Seat] = map[string]string{}
}
out[d.Seat][e.Local()] = e.SubjectPattern()
}
}
return out
}
// withEventSubjects gives each seat a declaration holds or watches the subjects of its events (machineEvent).
func withEventSubjects(subjects map[string]map[string]string, d *buscompose.Declared) {
if len(subjects) == 0 {
return
}
for _, list := range []*[]buscompose.Seat{&d.Holds, &d.Uses, &d.Watches} {
for i := range *list {
(*list)[i].EventSubjects = subjects[(*list)[i].Name]
}
}
}
+4 -2
View File
@@ -35,7 +35,7 @@ const (
)
// States is every state the part keeps.
var States = []string{StateAccounts, StateAssertions, StateGrants, StateMemberships, StateUserList, StateWithheld}
var States = []string{StateAccounts, StateAssertions, StateGrants, StateMemberships, StateRunning, StateUserList, StateWithheld}
// The events the part says.
const (
@@ -627,5 +627,7 @@ func (p *Part) inputs() Inputs {
p.cache.mu.Lock()
p.cache.gate = p.SeatList
p.cache.mu.Unlock()
return ReadInputs(p.Read, &p.cache)
in := ReadInputs(p.Read, &p.cache)
p.readRan(&in)
return in
}
@@ -44,7 +44,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
out := buscompose.Records{Assigned: map[string][]buscompose.Declared{}, People: map[string][]string{},
Interchangeable: map[string]bool{}, Enrolling: slices.Clone(in.Enrolling),
// And each node's logins (ADR 0583 §1): both while it rotates.
NodeLogins: in.NodeLogins, Kept: in.Kept,
NodeLogins: in.NodeLogins, Kept: in.Kept, EngineEvents: in.EventSubjects[buscompose.EngineSeat],
// What the bus module's process may define: every manifest's bus state, and what it said it created (DLV-G2).
BusState: map[string][]catalogue.BusStateEntry{}, Bound: in.Bound}
for module, m := range in.Manifests {
@@ -102,6 +102,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
// Its bindings as kept (bindings.go): the same source its membership's `bindings` is written from in this pass,
// so a grant and its binding are written together, never one before the other (issue 1966, the kind of issue 249).
d.Bindings = in.Bindings[m.Module]
withEventSubjects(in.EventSubjects, &d)
return d
}
d := declare(m)
@@ -132,6 +133,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
_, watches := consumedOf(m, seats, declarers)
out.Engine = &buscompose.Declared{Module: m.Module, Consumes: m.Consumes, Watches: watches, Invokes: m.Invokes,
State: bucketsFor(m, in.Nodes), Logs: logsOf(m), Streams: streamsOf(m)}
withEventSubjects(in.EventSubjects, out.Engine)
}
return out, gaps
}
@@ -439,7 +441,7 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
// holdersAfterPass is the holders on record (each node's assignments, as the seat mesh-module-assigner holds them) as they
// stand after catalogue.SeatsToPassOn (novox/hq ADR 0660): a holder whose every assignment's manifest no longer claims
// its seat has left it, and the seat is the only claimant's. Each assignment is read with the manifest its node runs
// (Inputs.Running), else the trunk's, else the running build of a held module; one with none is unread and keeps what it
// (Inputs.Ran: the digest its node-engine says it runs), else the trunk's, else the running build of a held module; one with none is unread and keeps what it
// holds. A pass is said as a gap and recorded nowhere (ADR 0660 §4). Pure.
func holdersAfterPass(in Inputs) ([]catalogue.Held, []Gap) {
var held []catalogue.Held
@@ -454,7 +456,7 @@ func holdersAfterPass(in Inputs) ([]catalogue.Held, []Gap) {
var assignments []catalogue.Assignment
for _, node := range slices.Sorted(maps.Keys(in.Assigned)) {
for _, module := range in.Assigned[node] {
m, ok := in.Running[node+"."+module]
m, ok := in.Ran[node+"."+module]
if !ok {
m, ok = in.Manifests[module]
}
@@ -43,7 +43,11 @@ const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent
// Consumes is every event the part hears, as its manifest's consumes name them.
// The seat mesh-broker's created, recreated and dropped are heard as every seat's event, `<seat>.<event>` on EVENTS (the
// lead's ruling, registry milestone 13:0x), never on the bus module's own stream.
var Consumes = append(append(slices.Clone(seatEvents), seatListEvents...), viewsEvents()...)
var Consumes = append(append(append(slices.Clone(seatEvents), seatListEvents...), viewsEvents()...), ConsumesRunning)
// ConsumesRunning is the seat node-engine's event running, in the local form the part's durable on its kept stream
// filters (running.go): it is heard on that stream alone, never on the shared events stream (SeatEventSubjects leaves it out).
const ConsumesRunning = "node-engine.running.*.*"
// seatListEvents are the two events of the seat mesh-catalog the seat list reader follows (mesh-sdk seatlist), heard by it on the
// module's event hub, so no handler of the hub takes them besides.
@@ -0,0 +1,116 @@
package grants
import (
"bytes"
"encoding/json"
"fmt"
"sort"
"strings"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// **What each node runs, as its node-engine says it** (novox/hq design 47, ADR 0660 §2): the seat node-engine's event
// `running`, kept last per subject on its own stream, is heard on the part's own durable and kept in the state `running`,
// one key `<node>.<module>` holding the event's body. holdersAfterPass reads the manifest of the digest each says, not the
// build the assigner wants.
// StateRunning is the part's state of what each node-engine said it runs.
const StateRunning = "running"
// RunningKey is a key of StateRunning.
func RunningKey(node, module string) string { return node + "." + module }
// HearRunning keeps one running event in the state running when it differs from the key's value (the time it was seen
// aside), and says whether the part must compose again. An event that is not a running body, or names a subject other
// than its own node and module, is refused with the reason; the caller drops it.
func (p *Part) HearRunning(subject string, data []byte) (changed bool, err error) {
var r nodeengine.RunningEvent
if err := json.Unmarshal(data, &r); err != nil {
return false, fmt.Errorf("%s is not a running body: %w", subject, err)
}
if r.Node == "" || r.Module == "" || r.Manifest == "" {
return false, fmt.Errorf("%s names no node, module or manifest digest", subject)
}
if !strings.HasSuffix(subject, "."+RunningKey(r.Node, r.Module)) {
return false, fmt.Errorf("%s does not carry the node and module its body names (%s)", subject, RunningKey(r.Node, r.Module))
}
key := RunningKey(r.Node, r.Module)
was, _, found, err := p.Store.Get(StateRunning, key)
if err != nil {
return false, fmt.Errorf("its own %s %s cannot be read: %w", StateRunning, key, err)
}
if found {
var w nodeengine.RunningEvent
if json.Unmarshal(was, &w) == nil && w.Manifest == r.Manifest && w.Fingerprint == r.Fingerprint {
return false, nil
}
}
if _, err := p.Store.Put(StateRunning, key, bytes.Clone(data)); err != nil {
return false, fmt.Errorf("its own %s %s cannot be written: %w", StateRunning, key, err)
}
return true, nil
}
// readRan reads each key of StateRunning and the manifest its digest names (the indexer's, by digest), into in.Ran. A key
// whose manifest cannot be read is a gap, and that node's module is read as it was before (the trunk's).
func (p *Part) readRan(in *Inputs) {
in.Ran = map[string]catalogue.Manifest{}
keys, err := p.Store.Keys(StateRunning)
if err != nil {
in.Gaps = append(in.Gaps, Gap{Input: StateRunning, Why: "its own state cannot be read, so no seat is read from what a node runs: " + err.Error()})
return
}
sort.Strings(keys)
for _, key := range keys {
raw, _, found, err := p.Store.Get(StateRunning, key)
if err != nil || !found {
continue
}
var r nodeengine.RunningEvent
if json.Unmarshal(raw, &r) != nil || r.Manifest == "" {
continue
}
m, err := p.cache.get(p.Read, r.Manifest)
if err != nil {
in.Gaps = append(in.Gaps, Gap{Input: "running manifest of " + key, Why: "the manifest the node-engine says it runs cannot be read, so the trunk's is used: " + err.Error()})
continue
}
in.Ran[key] = m
}
}
// RunningSaid is what the tool grants_running answers: each node and module the part has heard, the build and manifest
// digest, and when it was seen.
type RunningSaid struct {
Running []nodeengine.RunningEvent `json:"running"`
Note string `json:"note,omitempty"`
}
// RunningOf reads the state running, one key (`<node>.<module>`) or all.
func (p *Part) RunningOf(key string) (any, error) {
keys := []string{key}
if key == "" {
var err error
if keys, err = p.Store.Keys(StateRunning); err != nil {
return nil, err
}
sort.Strings(keys)
}
out := RunningSaid{Running: []nodeengine.RunningEvent{}}
for _, k := range keys {
raw, _, found, err := p.Store.Get(StateRunning, k)
if err != nil {
return nil, err
}
var r nodeengine.RunningEvent
if found && json.Unmarshal(raw, &r) == nil {
out.Running = append(out.Running, r)
}
}
if len(out.Running) == 0 {
out.Note = "no node-engine has said it runs anything that this part heard: seats are read from the trunk's manifests"
}
return out, nil
}
+3 -1
View File
@@ -28,6 +28,8 @@ const (
// ToolFirstViews and ToolFirstViewAgain read and ask again the part's first views (firstviews.go).
ToolFirstViews = "grants_first_views"
ToolFirstViewAgain = "grants_first_view_again"
// ToolRunning reads the state running (running.go).
ToolRunning = "grants_running"
// ToolSeatList and ToolSeatListAgain read and read again the seat list the part composes from (mesh-sdk seatlist.Tools).
ToolSeatList = "grants_seat_list"
ToolSeatListAgain = "grants_seat_list_again"
@@ -35,7 +37,7 @@ const (
// Tools is every tool, in the manifest's order.
var Tools = []string{ToolGrants, ToolMembership, ToolUserList, ToolComposeNow, ToolAssert, ToolAccounts, ToolConsumerMatch,
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolSeatList, ToolSeatListAgain}
ToolAcceptHashes, ToolFirstViews, ToolFirstViewAgain, ToolRunning, ToolSeatList, ToolSeatListAgain}
// WritesGrants is what grants says of the part's own grants: written once the list the bus loaded carries a build's
// accounts, and the one key the assigner reads (ADR 0571 clause 9, which changes ADR 0547 clause 5).
+7 -1
View File
@@ -24,7 +24,8 @@
"mesh-catalog.manifest-accepted",
"mesh-catalog.manifest-refused",
"mesh-catalog.seat-definition-changed",
"mesh-catalog.seat-definition-removed"
"mesh-catalog.seat-definition-removed",
"node-engine.running.*.*"
],
"uses": [
"operator-channel"
@@ -61,6 +62,7 @@
"grants_accept_hashes",
"grants_first_views",
"grants_first_view_again",
"grants_running",
"grants_seat_list",
"grants_seat_list_again"
],
@@ -81,6 +83,10 @@
"name": "memberships",
"history": 5
},
{
"name": "running",
"history": 1
},
{
"name": "user-list",
"history": 20