From 4a61381ff1bae7d653cb4b8ad303b2a568179347 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 21:18:15 +0100 Subject: [PATCH] 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 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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..*) and its durable on the kept stream by name. Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ Builds: design 47 G §199, §249, §309 --- .../cmd/mesh-grants/integration_test.go | 39 ++++++ modules/mesh-grants/cmd/mesh-grants/main.go | 37 ++++++ modules/mesh-grants/cmd/mesh-grants/tools.go | 6 + modules/mesh-grants/grants_test.go | 48 ++++++++ .../internal/buscompose/buscompose_test.go | 43 +++++++ .../internal/buscompose/busstate.go | 2 +- .../mesh-grants/internal/buscompose/engine.go | 25 ++++ .../mesh-grants/internal/buscompose/nats.go | 52 ++++++-- .../mesh-grants/internal/buscompose/users.go | 16 ++- modules/mesh-grants/internal/grants/follow.go | 14 ++- modules/mesh-grants/internal/grants/inputs.go | 7 ++ modules/mesh-grants/internal/grants/kept.go | 30 +++++ modules/mesh-grants/internal/grants/part.go | 6 +- .../mesh-grants/internal/grants/records.go | 8 +- .../mesh-grants/internal/grants/rootfree.go | 6 +- .../mesh-grants/internal/grants/running.go | 116 ++++++++++++++++++ modules/mesh-grants/internal/grants/tools.go | 4 +- modules/mesh-grants/module.json | 8 +- 18 files changed, 447 insertions(+), 20 deletions(-) create mode 100644 modules/mesh-grants/internal/grants/running.go diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index b48e8e51..3d2e578b 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -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 { diff --git a/modules/mesh-grants/cmd/mesh-grants/main.go b/modules/mesh-grants/cmd/mesh-grants/main.go index 67114a96..d7d36e14 100644 --- a/modules/mesh-grants/cmd/mesh-grants/main.go +++ b/modules/mesh-grants/cmd/mesh-grants/main.go @@ -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 +} diff --git a/modules/mesh-grants/cmd/mesh-grants/tools.go b/modules/mesh-grants/cmd/mesh-grants/tools.go index 954b8f47..9d68a0a6 100644 --- a/modules/mesh-grants/cmd/mesh-grants/tools.go +++ b/modules/mesh-grants/cmd/mesh-grants/tools.go @@ -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, `.`, or all. Read-only.", + Input: schema(map[string]string{"key": "one `.`; none for all"}), + Run: func(a map[string]any) (any, error) { return p.RunningOf(grants.Answer(a, "key")) }}, } } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index e4a02217..cf70a3e7 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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 /"}, } + // 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) + } +} diff --git a/modules/mesh-grants/internal/buscompose/buscompose_test.go b/modules/mesh-grants/internal/buscompose/buscompose_test.go index 157e7c9e..c6cc96e7 100644 --- a/modules/mesh-grants/internal/buscompose/buscompose_test.go +++ b/modules/mesh-grants/internal/buscompose/buscompose_test.go @@ -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 ``: 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..", "composed.*": "composed.", "owed.*": "owed."}} + for _, row := range []struct { + name, event, want string + ok bool + }{ + {"the node is the first part", "running.*.*", "running..*", true}, + {"the node is the only part", "composed.*", "composed.", true}, + {"a part that is no node's says no machine", "owed.*", "", false}, + {"an event with no subject takes the node last", "done", "done.", 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) + } +} diff --git a/modules/mesh-grants/internal/buscompose/busstate.go b/modules/mesh-grants/internal/buscompose/busstate.go index 86d9bf7d..f0716943 100644 --- a/modules/mesh-grants/internal/buscompose/busstate.go +++ b/modules/mesh-grants/internal/buscompose/busstate.go @@ -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)) } diff --git a/modules/mesh-grants/internal/buscompose/engine.go b/modules/mesh-grants/internal/buscompose/engine.go index 236e5bd1..3dadd65b 100644 --- a/modules/mesh-grants/internal/buscompose/engine.go +++ b/modules/mesh-grants/internal/buscompose/engine.go @@ -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 `` (design 47: + // `running..`): 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)) diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index 40f16ddb..e21e5807 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -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..`): 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.`, 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 = "" + +// 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 `` 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 `` (Seat.EventSubjects: +// `running..`), 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)) } diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index e7f2a49f..0bfb2456 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -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 +} diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index 1344e273..7ba13912 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -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 { diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index ae44400a..4904bb14 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -155,6 +155,9 @@ type Inputs struct { // Running is the manifest a machine is wanted to run, by `.`: 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 `.` (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 // `.`; 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 diff --git a/modules/mesh-grants/internal/grants/kept.go b/modules/mesh-grants/internal/grants/kept.go index 31e03251..79cdea89 100644 --- a/modules/mesh-grants/internal/grants/kept.go +++ b/modules/mesh-grants/internal/grants/kept.go @@ -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..`) 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] + } + } +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index 3edf13f7..85f9b9fa 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -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 } diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index 36c611bb..c20ef223 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -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] } diff --git a/modules/mesh-grants/internal/grants/rootfree.go b/modules/mesh-grants/internal/grants/rootfree.go index 72931a69..36aacd36 100644 --- a/modules/mesh-grants/internal/grants/rootfree.go +++ b/modules/mesh-grants/internal/grants/rootfree.go @@ -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, `.` 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. diff --git a/modules/mesh-grants/internal/grants/running.go b/modules/mesh-grants/internal/grants/running.go new file mode 100644 index 00000000..8cda4a51 --- /dev/null +++ b/modules/mesh-grants/internal/grants/running.go @@ -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 `.` 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 (`.`) 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 +} diff --git a/modules/mesh-grants/internal/grants/tools.go b/modules/mesh-grants/internal/grants/tools.go index 31f46c8d..07ef30c7 100644 --- a/modules/mesh-grants/internal/grants/tools.go +++ b/modules/mesh-grants/internal/grants/tools.go @@ -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). diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 4ae5ec87..6443652e 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -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