Files
mesh-delivery/modules/mesh-grants/internal/grants/records.go
T
jschoubben 3b6f3af38e Grants part: create a module's log with the max age it declares, so a log's entries age out as its manifest says (issue 1952)
Its nats fixture gives the tools bundle's account a secret the tool runtime can reach, as mesh-sdk 2e5cc4f asks. Pins mesh-sdk 2e5cc4f.
2026-10-29 22:19:35 +01:00

419 lines
16 KiB
Go

package grants
import (
"fmt"
"reflect"
"slices"
"strings"
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Records is what the bus composer composes from, rebuilt from the bus (mesh-controller internal/inventory busrecords.go,
// BusRecords, read the same way from Inputs instead of the controller's store). Pure. An assigned module whose manifest
// the indexer does not hold is left out and named, where the controller refuses the whole composition: the shadow
// composes what it can and says what it could not.
func Records(in Inputs) (buscompose.Records, []Gap) { return records(in, false) }
// ListRecords is Records for the user list and the grants (issue 774): a module whose wanted build's manifest is not the
// trunk's is declared with what both grant (unionDeclared), so the list carries the trunk manifest's subjects before the
// assigner moves the want to it, and a grant at the trunk's fingerprint waits for nothing the want's move would bring.
// Memberships and consumers stay with Records: they say what runs. Pure.
func ListRecords(in Inputs) (buscompose.Records, []Gap) { return records(in, true) }
func records(in Inputs, union bool) (buscompose.Records, []Gap) {
var gaps []Gap
seats := map[string]catalogue.SeatDeclaration{}
declarers := map[string]string{}
for _, m := range in.Manifests {
for _, s := range m.DefinesSeats {
seats[s.Name] = s
declarers[s.Name] = m.Module
}
}
for _, own := range catalogue.SeatsWithAProtocol() {
seats[own.Name] = catalogue.SeatDeclaration{Name: own.Name, Scope: own.Scope, Accepts: own.Accepts, Emits: own.Emits, Serves: own.Serves}
}
out := buscompose.Records{Assigned: map[string][]buscompose.Declared{}, People: map[string][]string{},
Interchangeable: map[string]bool{}}
values := func(provider, setting string) []any {
var vs []any
for _, node := range in.Nodes {
if v, ok := in.Values[node+"/"+provider+"/"+setting]; ok && slices.Contains(in.Assigned[node], provider) {
vs = append(vs, v)
}
}
return vs
}
for _, node := range in.Nodes {
out.Nodes = append(out.Nodes, node)
manifestOf := func(module string) (catalogue.Manifest, bool) {
if m, ok := in.Running[node+"."+module]; ok {
return m, true
}
m, ok := in.Manifests[module]
return m, ok
}
var here []catalogue.Manifest
for _, module := range in.Assigned[node] {
if m, ok := manifestOf(module); ok {
here = append(here, m)
}
}
for _, module := range in.Assigned[node] {
m, known := manifestOf(module)
if !known {
gaps = append(gaps, Gap{Input: module + " on " + node, Why: "left out of the shadow list: no manifest to derive what it may say from"})
continue
}
declare := func(m catalogue.Manifest) buscompose.Declared {
d := declaredFor(m, seats, declarers)
if len(in.Placed) > 0 {
d.Holds = heldWherePlaced(d.Holds, in.Placed, node, m.Module)
}
for _, sr := range catalogue.ReadsGranted(m, here, node) {
d.KeyedReads = append(d.KeyedReads, buscompose.KeyedRead{Read: sr.Read, Key: sr.Key})
}
d.Hears, d.HearsAStream, d.Consumes = hearsOf(m, in.Manifests, node, values, d.Consumes)
return d
}
d := declare(m)
trunk, onTrunk := in.Manifests[module]
// What a node composing a consumer reads of it is the trunk manifest's served keys (frombus reads the indexer's
// manifest of another node's module), and what runs may still be another build: both are granted (issue 1126).
if onTrunk {
d.Served = slices.Compact(slices.Sorted(slices.Values(append(slices.Clone(d.Served), catalogue.ServedSettings(trunk)...))))
}
if union && onTrunk && !reflect.DeepEqual(m, trunk) {
d = unionDeclared(d, declare(trunk))
}
out.Assigned[node] = append(out.Assigned[node], d)
if m.Instances == catalogue.InstancesInterchangeable || union && onTrunk && trunk.Instances == catalogue.InstancesInterchangeable {
out.Interchangeable[m.Module] = true
}
}
}
// The node-engine's own manifest, placed on every node (ADR 0563): its state, logs and streams alone, created by this
// part; it adds no principal, since each node's engine speaks as the node's account.
if m, ok := in.Manifests[buscompose.EngineModule]; ok {
out.Engine = &buscompose.Declared{Module: m.Module, State: bucketsOf(m), Logs: logsOf(m), Streams: streamsOf(m)}
}
return out, gaps
}
// declaredFor is mesh-controller's declaredFor (internal/inventory/busrecords.go), copied: one module's manifest as the
// composer needs it.
func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration, declarers map[string]string) buscompose.Declared {
var fromModules []string
var watches []buscompose.Seat
for _, c := range m.Consumes {
emitter, event, named := strings.Cut(c, ".")
if named {
if s, isASeat := seats[emitter]; isASeat && s.Kinded {
ev := strings.TrimSuffix(event, ".*")
if catalogue.SeatSays(s.Emits, ev) {
watches = append(watches, buscompose.Seat{Name: s.Name, Scope: s.Scope, Emits: []string{ev},
Kinded: true, Proofs: s.Proofs, DeclaredBy: declarers[s.Name]})
continue
}
}
if s, isASeat := seats[emitter]; isASeat && catalogue.SeatSays(s.Emits, event) {
watches = append(watches, buscompose.Seat{Name: s.Name, Emits: []string{event}})
continue
}
}
fromModules = append(fromModules, c)
}
d := buscompose.Declared{
Module: m.Module, Emits: m.EmitsAll(), Consumes: fromModules, Watches: watches, Serves: m.Tools,
Invokes: m.Invokes, Waits: waitsOf(m, seats), State: bucketsOf(m), Reads: m.Reads, Logs: logsOf(m),
Streams: streamsOf(m), Queues: queueOwnersOf(m), Processes: processesOf(m), Checks: catalogue.HealthChecks(m),
RunsAs: m.RunsAs, Issued: issuedTools(m),
// The keys its served values fill, which every node may read where it runs (issue 1126).
Served: catalogue.ServedSettings(m),
}
d.Loads = loadsABundle(m)
if _, reads := m.OwnSecrets["broker"]; !reads {
d.NoAccount = true
}
if _, reads := m.OwnSecrets[buscompose.ProvisionerSecret]; reads && m.ClaimsSeat(buscompose.NodeProvisionerSeat) {
d.ProvisionsNode = true
}
d.SnapshotsTheBus = m.ClaimsSeat(catalogue.BrokerSeat)
for _, c := range m.Claims {
if s, hasAProtocol := seats[c.Name]; hasAProtocol {
held := asSeat(s, declarers[s.Name])
held.Kind = c.Kind
held.Capabilities = c.Capabilities
d.Holds = append(d.Holds, held)
}
}
for _, name := range m.Uses {
if s, declaredSomewhere := seats[name]; declaredSomewhere {
d.Uses = append(d.Uses, asSeat(s, declarers[s.Name]))
}
}
return d
}
// loadsABundle says the node's tool runtime loads a bundle of the module. The controller reads it from the resolved
// manifest's Bundles, which a build derives and the indexer's manifest does not carry (the builder's build record names
// the repository's manifest digest), so the same answer is derived here from `build.artifacts` by mesh-sdk's rule
// (the build's own rule for what a bundle loads): a bundle that says `loads`, else every bundle of a module that declares tools and that no
// resource of the module runs. Without it a module whose only runtime-loaded part is a helper (route-proxy's
// route-provider) was left out of node-tools' grants (issue 1945).
func loadsABundle(m catalogue.Manifest) bool {
for _, b := range m.Bundles {
if len(b.Loads) > 0 {
return true
}
}
if m.Build == nil {
return false
}
run := map[string]bool{}
for _, r := range m.Resources {
if a, ok := r["artifact"].(string); ok && a != "" {
run[a] = true
}
}
for _, a := range m.Build.Artifacts {
if a.Kind != catalogue.ArtifactBundle {
continue
}
if len(a.Loads) > 0 || (a.Loads == nil && len(m.Tools) > 0 && !run[a.Name]) {
return true
}
}
return false
}
func waitsOf(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaration) []buscompose.Wait {
var out []buscompose.Wait
for _, w := range m.Waits {
emitter, event, _ := strings.Cut(w, ".")
first, _, _ := strings.Cut(event, ".")
wait := buscompose.Wait{Name: w, Emitter: emitter, Event: event}
if s, isASeat := seats[emitter]; isASeat {
switch {
case s.Kinded && catalogue.SeatSays(s.Emits, first):
wait.Seat, wait.Last = true, "kind"
case s.NamedByCaller(first) && catalogue.SeatSays(s.Emits, first):
wait.Seat, wait.Last = true, "asker"
case catalogue.SeatSays(s.Emits, event):
wait.Seat = true
}
}
out = append(out, wait)
}
return out
}
func issuedTools(m catalogue.Manifest) []string {
var out []string
for _, o := range m.Provides {
if tool, issued := m.IssuedCredentialOf(o.Name); issued && !slices.Contains(out, tool) {
out = append(out, tool)
}
}
return out
}
func bucketsOf(m catalogue.Manifest) []buscompose.Bucket {
var out []buscompose.Bucket
for _, s := range m.State {
out = append(out, buscompose.Bucket{Module: m.Module, Name: s.Name, History: s.History, TTLSeconds: s.TTLSeconds,
PerMachine: s.PerMachine, KeyTTL: s.KeyTTL})
}
return out
}
func logsOf(m catalogue.Manifest) []buscompose.Log {
var out []buscompose.Log
for _, l := range m.Logs {
out = append(out, buscompose.Log{Module: m.Module, Name: l.Name, MaxMiB: l.Cap(), MaxAgeDays: l.MaxAgeDays})
}
return out
}
func asSeat(s catalogue.SeatDeclaration, declarer string) buscompose.Seat {
seat := buscompose.Seat{Name: s.Name, Scope: s.Scope, Accepts: s.Accepts, Emits: s.Emits,
Serves: catalogue.VerbNames(s.Serves), Kinded: s.Kinded, ByCaller: s.ByCaller, Proofs: s.Proofs, DeclaredBy: declarer}
if own, ok := catalogue.SeatNamed(s.Name); ok && own.PulledByKind {
seat.PulledByKind = true
}
for _, r := range s.Records {
if declarer != "" {
seat.Records = append(seat.Records, buscompose.BucketName(declarer, r))
}
}
return seat
}
func streamsOf(m catalogue.Manifest) []buscompose.ModuleStream {
var out []buscompose.ModuleStream
for _, s := range m.Streams {
events := make([]string, 0, len(s.Events))
for _, e := range s.Events {
events = append(events, buscompose.WirePattern(e))
}
out = append(out, buscompose.ModuleStream{Module: m.Module, Name: s.BusName(), Events: events,
MaxAgeDays: s.Days(), MaxMiB: s.Cap(), Restricted: s.Grant != nil,
WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates})
}
return out
}
func queueOwnersOf(m catalogue.Manifest) []buscompose.QueueOwner {
buses := m.ProcessBuses()
if len(buses) == 0 {
return nil
}
byName := map[string]catalogue.StreamDeclaration{}
for _, s := range m.Streams {
byName[s.Name] = s
}
var out []buscompose.QueueOwner
for _, r := range m.Resources {
id := fmt.Sprint(r["id"])
b, declared := buses[id]
if !declared {
continue
}
for _, q := range b.Queues {
s, known := byName[q]
if !known || s.Retention != catalogue.StreamRetentionWorkQueue {
continue
}
out = append(out, buscompose.QueueOwner{Process: id, Stream: s.BusName(), MaxDeliver: s.MaxDeliveries})
}
}
return out
}
func processesOf(m catalogue.Manifest) []buscompose.ProcessPrincipal {
accounts := m.BusAccounts()
if len(accounts) == 0 {
return nil
}
queues := queueOwnersOf(m)
var out []buscompose.ProcessPrincipal
// Every own bus account, processes first and then tools bundles (catalogue.BusAccounts): a tools bundle's account carries
// what it may remove from the bus (ADR 0564 Limits 4, issue 1952 S5).
for _, a := range accounts {
pr := buscompose.ProcessPrincipal{Process: a.ID, Emits: a.Bus.Emits, State: a.Bus.State, Asks: a.Bus.Asks,
Invokes: a.Bus.Invokes, Removes: a.Bus.Removes}
for _, q := range queues {
if q.Process == a.ID {
pr.Queues = append(pr.Queues, q)
}
}
out = append(out, pr)
}
return out
}
func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node string, values func(provider, setting string) []any,
consumes []string) ([]buscompose.Heard, bool, []string) {
reading := m
reading.Consumes = consumes
parts, touches, onEvents := catalogue.Hearing(reading, declared, node, values)
var out []buscompose.Heard
for _, p := range parts {
h := buscompose.Heard{Stream: p.Stream.BusName(), Emitter: p.Emitter, Restricted: p.Stream.Grant != nil}
for _, pattern := range p.Patterns {
h.Filters = append(h.Filters, buscompose.ModuleStreamSubject(p.Emitter, buscompose.WirePattern(pattern)))
}
out = append(out, h)
}
return out, touches, onEvents
}
// heldWherePlaced is a module's holds on a node as the catalogue's placement says them (ADR 0549 §1, §3). A placement's
// holds are the holders on record, as catalogue.World's Holdings are: a seat held once for the mesh that some node's
// placement holds is held only where a placement names it (this node's naming this module), so a module claiming it on
// two machines holds it on the one placement names, not on both. A seat no placement holds is held by derivation, by the
// module claiming it wherever it is placed, as it was before placement: the seeded placements hold only the seats the
// controller had on record (twelve on the control-node), and treating their holds as every holder left nine seats with
// no holder in the user list of 2026-10-27 09:21 (issue 1945). A seat of every node is held wherever the module runs.
func heldWherePlaced(holds []buscompose.Seat, placed map[string]delivery.Placement, node, module string) []buscompose.Seat {
same := func(x delivery.Holding, h buscompose.Seat) bool {
return (x.Scope == "" || x.Scope == catalogue.ScopeMesh) && sameSeat(x.Claim, h.Name) && x.Kind == h.Kind
}
var out []buscompose.Seat
for _, h := range holds {
if h.Scope != "" && h.Scope != catalogue.ScopeMesh {
out = append(out, h)
continue
}
onRecord := false
for _, p := range placed {
if slices.ContainsFunc(p.Holds, func(x delivery.Holding) bool { return same(x, h) }) {
onRecord = true
break
}
}
if !onRecord || slices.ContainsFunc(placed[node].Holds, func(x delivery.Holding) bool { return same(x, h) && x.Module == module }) {
out = append(out, h)
}
}
return out
}
// sameSeat says two claims name one seat, under either of its names (catalogue.SeatNamed, ADR 0122).
func sameSeat(a, b string) bool {
if a == b {
return true
}
sa, oka := catalogue.SeatNamed(a)
sb, okb := catalogue.SeatNamed(b)
return oka && okb && sa.Name == sb.Name
}
// unionDeclared is what either declaration grants (issue 774): every list of b's entries a lacks appended to a's, a
// process of both merged by its name, a flag that adds a grant set where either sets it, and an account kept where
// either has one (NoAccount only where both say none). a's names and RunsAs stand. Pure.
func unionDeclared(a, b buscompose.Declared) buscompose.Declared {
out := a
unionFields(reflect.ValueOf(&out).Elem(), reflect.ValueOf(b))
out.NoAccount = a.NoAccount && b.NoAccount
procs := slices.Clone(a.Processes)
for _, bp := range b.Processes {
i := slices.IndexFunc(procs, func(ap buscompose.ProcessPrincipal) bool { return ap.Process == bp.Process })
if i < 0 {
procs = append(procs, bp)
continue
}
merged := procs[i]
unionFields(reflect.ValueOf(&merged).Elem(), reflect.ValueOf(bp))
procs[i] = merged
}
out.Processes = procs
return out
}
// unionFields appends to each slice field of dst the elements of src's it lacks, and sets each bool field src sets.
func unionFields(dst, src reflect.Value) {
for i := 0; i < dst.NumField(); i++ {
f, g := dst.Field(i), src.Field(i)
switch f.Kind() {
case reflect.Slice:
merged := reflect.AppendSlice(reflect.MakeSlice(f.Type(), 0, f.Len()+g.Len()), f)
for j := 0; j < g.Len(); j++ {
held := false
for k := 0; k < merged.Len() && !held; k++ {
held = reflect.DeepEqual(merged.Index(k).Interface(), g.Index(j).Interface())
}
if !held {
merged = reflect.Append(merged, g.Index(j))
}
}
f.Set(merged)
case reflect.Bool:
f.SetBool(f.Bool() || g.Bool())
}
}
}