mesh-grants: create a module's object buckets and history streams, grant their readers read-only, and warn at 75 percent

ADR 0592 §5, §6, §9, §25 (the grants part's share by ADR 0593 §10): an object bucket and a stream that
keeps its history are created create-only with their caps and no age limit, refusing new data when full;
the owner alone writes its bucket and a module its grant names reads it; the part raises
mesh-grants.history-full once any passes 75 percent. Each engine also reads the indexer's manifest
verdicts by one key for compose-candidate.

Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
2026-11-02 18:56:19 +01:00
parent d283efdc9f
commit 0b03b9c7c5
14 changed files with 444 additions and 21 deletions
@@ -33,7 +33,8 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
"novox": {{Module: "nats", SnapshotsTheBus: true},
{Module: buscompose.GrantsModule, State: []buscompose.Bucket{{Module: buscompose.GrantsModule, Name: "user-list"}}}},
"ace": {{Module: "plex", Reads: []string{delivery.ComposedRead}},
{Module: "mesh-delivery-view", Reads: []string{delivery.ComposedRead}}},
{Module: "mesh-delivery-view", Reads: []string{delivery.ComposedRead}, ReadsObjects: []string{"mesh-knowledge_sections"}},
{Module: "mesh-knowledge", Objects: []buscompose.ObjectBucket{{Module: "mesh-knowledge", Name: "sections", MaxMiB: 1, Restricted: true}}}},
},
}
// The node-engine's state as the part's records carry it: every node's compositions (ADR 0593 §6).
@@ -216,6 +217,9 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
if _, err := other.GetLastMsg("KV_"+buscompose.GrantsUserListBucket, "$KV."+buscompose.GrantsUserListBucket+"."+buscompose.GrantsUserListKey, nats.DirectGet()); err == nil {
t.Errorf("another node read the key: %v", err)
}
// A context of its own: the refusals above each wait out their bound.
ctx, cancel = context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// ADR 0593 §6 (D14): the grants part creates each node's compositions; ace's engine writes its own and not novox's;
// the view, a named reader, finds every node's bucket by listing and reads it, writing nothing; plex reads none.
for _, node := range records.Nodes {
@@ -249,11 +253,48 @@ func theBusNodeFollowsTheSealedListUnderTheComposedGrants(t *testing.T) {
if _, err := composed.Get(ctx, "ace.node"); err != nil {
t.Errorf("the view reads ace's own composition: %v", err)
}
if _, err := composed.Put(ctx, "ace.node", []byte(`{}`)); err == nil {
refusedIn := func() context.Context { // a refused publish is answered by no one: it waits out a short bound
c, stop := context.WithTimeout(ctx, 2*time.Second)
t.Cleanup(stop)
return c
}
if _, err := composed.Put(refusedIn(), "ace.node", []byte(`{}`)); err == nil {
t.Error("the view wrote a composition")
}
plex, _ := dial("ace.plex").JetStream(nats.MaxWait(2 * time.Second))
if _, err := plex.GetLastMsg("KV_composed-ace", "$KV.composed-ace.ace.node", nats.DirectGet()); err == nil {
t.Error("plex, not a named reader, read a composition")
}
// ADR 0592 §5, §9: a module's own object bucket, created by the grants part, written (and an object replaced) by its
// owner alone, read by the module its grant names, read-only, and by no other.
sections := buscompose.ObjectBucket{Module: "mesh-knowledge", Name: "sections", MaxMiB: 1, Restricted: true}
if _, err := grants.CreateStream(ctx, buscompose.ObjectsStreamConfig(sections)); err != nil {
t.Fatalf("the grants part creates %s: %v", sections.Stream(), err)
}
owner, err := jetstream.New(dial("ace.mesh-knowledge"))
if err != nil {
t.Fatal(err)
}
ownStore, err := owner.ObjectStore(ctx, sections.Bucket())
if err != nil {
t.Fatalf("the owner binds its object bucket: %v", err)
}
for range 2 { // put twice: a second put of a name purges its old chunks
if _, err := ownStore.PutBytes(ctx, "s1", []byte("body")); err != nil {
t.Fatalf("the owner puts an object: %v", err)
}
}
readStore, err := view.ObjectStore(ctx, sections.Bucket())
if err != nil {
t.Fatalf("the granted reader binds the bucket: %v", err)
}
if got, err := readStore.GetBytes(ctx, "s1"); err != nil || string(got) != "body" {
t.Errorf("the granted reader reads %q, %v", got, err)
}
if _, err := readStore.PutBytes(refusedIn(), "s2", []byte("x")); err == nil {
t.Error("the granted reader wrote the bucket")
}
if _, err := plex.GetLastMsg(sections.Stream(), "$O."+sections.Bucket()+".M.x", nats.DirectGet()); err == nil {
t.Error("a module the grant does not name read the bucket")
}
}
+79 -1
View File
@@ -633,6 +633,84 @@ func TestGrants(t *testing.T) {
}
})
// ADR 0592 §5, §6, §9 (ADR 0593 §10): a module's own object bucket and a stream that keeps its history are created
// by the part, create-only, with their caps; the owner alone writes the bucket; a module its trusted value grants reads
// it read-only; a module its reads name and the grant does not is refused it.
t.Run("objects: a module's object bucket and history stream are created, written by it alone, read by its grant", func(t *testing.T) {
const knowledge = `{"module":"mesh-knowledge","version":"1","runs-as":"k","own-secrets":{"broker":"/var/lib/k/broker"},"secrets-owner":"k",` +
`"streams":[{"name":"versions","events":["version.**"],"max-mib":2048,"retention":"history"}],` +
`"objects":[{"name":"sections","max-mib":4096,"grant":{"config":"readers"}}],"resources":[{"id":"state","type":"directory","mode":"0700","place":"."},` +
`{"id":"readers","trusted":true,"type":"file","path":"${dir:state}/readers.json","mode":"0600","merge":"json","takes-settings":["readers"],"content":"{\"readers\": {}}"}]}`
b := fixture()
b.manifestOf("mesh-knowledge", knowledge)
for _, m := range []string{"web", "other"} {
b.manifestOf(m, `{"module":"`+m+`","version":"1","own-secrets":{"broker":"/var/lib/`+m+`/broker"},"secrets-owner":"`+m+`","reads":["mesh-knowledge.sections"]}`)
b.assign("a", m)
}
b.assign("a", "mesh-knowledge")
b.put(grants.ReadResolved, "a.mesh-knowledge.readers", map[string]any{"node": "a", "module": "mesh-knowledge", "key": "readers",
"value": map[string]any{"records": []string{"web"}}})
b.loaded = shadowOf(t, b)
p := partOn(b)
p.Create = b
p.Compose()
km, err := catalogue.Decode([]byte(knowledge))
if err != nil {
t.Fatal(err)
}
cfgs := grants.Configs(km, nil)
objects, versions := cfgs["OBJ_mesh-knowledge_sections"], cfgs["VERSIONS"]
if !slices.Contains(b.creates, "OBJ_mesh-knowledge_sections") || !slices.Contains(b.creates, "VERSIONS") || objects.MaxBytes != 4096<<20 || objects.MaxAge != 0 ||
objects.Discard != jetstream.DiscardNew {
t.Errorf("the object bucket: created %v, as %+v", b.creates, objects)
}
if versions.MaxAge != 0 || versions.Discard != jetstream.DiscardNew || versions.MaxBytes != 2048<<20 {
t.Errorf("the history stream is composed as %+v (of %v)", versions, slices.Sorted(maps.Keys(cfgs)))
}
users := shadowOf(t, b)
of := func(name string) grants.User {
return users[slices.IndexFunc(users, func(u grants.User) bool { return u.User == name })]
}
owner, web, other := of("a.mesh-knowledge"), of("a.web"), of("a.other")
const read = "$JS.API.DIRECT.GET.OBJ_mesh-knowledge_sections.>"
if !slices.Contains(owner.Publish, "$O.mesh-knowledge_sections.>") || !slices.Contains(web.Publish, read) ||
slices.Contains(web.Publish, "$O.mesh-knowledge_sections.>") || slices.Contains(other.Publish, read) {
t.Errorf("the owner writes %v; web reads %v; other reads %v", owner.Publish, web.Publish, other.Publish)
}
if slices.Contains(web.Publish, "$JS.API.DIRECT.GET.KV_mesh-knowledge_sections.>") {
t.Error("the object bucket's read is granted as a state's")
}
// ADR 0592 §6, §25: the bucket past 75 percent of its cap is said, in one warning, and cleared once below; the history
// stream at 10 percent is not.
var said []asks.Condition
p.Say = func(c asks.Condition) error {
if c.Key == grants.FullCondition {
said = append(said, c)
}
return nil
}
size := func(stream string, bytes int64) {
for _, s := range b.streams {
if s["name"] == stream {
s["bytes"] = bytes
}
}
}
size("OBJ_mesh-knowledge_sections", 3500<<20)
size("VERSIONS", 200<<20)
if res := p.Compose(); len(res.Full) != 1 || res.Full[0].Stream != "OBJ_mesh-knowledge_sections" || res.Full[0].Percent() != 85 ||
len(said) != 1 || said[0].State != asks.StateRaised || !strings.Contains(said[0].Summary, "3500 of 4096 MiB (85%)") {
t.Fatalf("full = %+v, said %+v", res.Full, said)
}
if p.Compose(); len(said) != 1 {
t.Errorf("the same fill was said again: %+v", said)
}
size("OBJ_mesh-knowledge_sections", 1<<20)
if res := p.Compose(); len(res.Full) != 0 || len(said) != 2 || said[1].State != asks.StateCleared {
t.Errorf("below its mark: full %+v, said %+v", res.Full, said)
}
})
t.Run("withheld: a hearing of a restricted stream in the tool runner is written on change only, and deleted once heard (issue 1955 L7)", func(t *testing.T) {
const mail = `{"module":"mesh-mail","version":"1","streams":[{"name":"mail","events":["mail.*.message.added"],"max-age-days":7,` +
`"grant":{"config":"grants","events":"mail.{key}.**"}}],"resources":[{"id":"state","type":"directory","mode":"0700","place":"."},` +
@@ -2291,7 +2369,7 @@ func TestGrants(t *testing.T) {
var said, behind []asks.Condition
p.Say = func(c asks.Condition) error {
switch c.Key {
case grants.GrantMissingCondition, grants.RefusedCondition: // their own rows
case grants.GrantMissingCondition, grants.RefusedCondition, grants.FullCondition: // their own rows
case grants.ConsumerBehindCondition:
behind = append(behind, c)
default:
@@ -29,11 +29,14 @@ func ComposedReads(node string) ([]string, error) {
return append(out, CandidateReads...), nil
}
// CandidateReads is what a node's engine is granted of the indexer's candidates to answer `compose-candidate` (ADR 0593
// §7): binding the bucket and a direct get of one key; no consumer, so it neither lists nor watches them.
// CandidateReads is what a node's engine is granted of the indexer's candidates and manifest verdicts to answer
// `compose-candidate` (ADR 0593 §7, ADR 0578 §3): binding each bucket and a direct get of one key; no consumer, so it
// neither lists nor watches them.
var CandidateReads = []string{
"$JS.API.DIRECT.GET.KV_" + delivery.CandidatesBucket + ".>",
"$JS.API.STREAM.INFO.KV_" + delivery.CandidatesBucket,
"$JS.API.DIRECT.GET.KV_" + delivery.VerdictsBucket + ".>",
"$JS.API.STREAM.INFO.KV_" + delivery.VerdictsBucket,
}
// EngineComposedBuckets are the composition buckets of every node, as the node-engine's state (ADR 0593 §6): the grants
@@ -55,6 +55,9 @@ type ModuleStream struct {
// Duplicates is the work-queue's duplicate window in hours (novox/hq ADR 0436 §6); meaningless unless
// WorkQueue.
Duplicates int
// History says the stream keeps its history (novox/hq ADR 0592 §6, `retention: history`): every message, with no
// maximum age, a new one refused when it is full rather than its oldest dropped.
History bool
}
// Subjects are the subjects the stream takes.
@@ -90,6 +93,15 @@ func (s ModuleStream) Stream() Stream {
// Said on the bus, so a dead letter kept from it ages out with it (novox/hq ADR 0376 §3).
Restricted: s.Restricted,
}
if s.History {
out.MaxAge = 0
out.DiscardNew = true
out.Why = fmt.Sprintf("%s's own stream keeping its history (novox/hq ADR 0592 §6): every message, with no maximum "+
"age, at most %d MiB, a new message refused when full", s.Module, s.MaxMiB)
if s.Restricted {
out.Why += "; heard only by the consumers its trusted config value grants"
}
}
if s.WorkQueue {
out.Retention = RetentionWorkQueue
out.DiscardNew = true
@@ -106,6 +106,9 @@ type Principal struct {
// Username and inbox are <node>.<module>.<process id>.
Process string
// Objects are the bus names of its module's own object buckets, and ReadsObjects those of others it reads (novox/hq
// ADR 0592 §5, §9).
Objects, ReadsObjects []string
// Nodes are every node the records know, for a read of a state kept one bucket per node (delivery.ComposedRead, ADR
// 0593 §6): set only where the principal or a module it carries reads one.
Nodes []string
@@ -929,6 +932,8 @@ func PermissionsFor(p Principal) (Permissions, error) {
// And it hears each of those buckets created, so one not on the bus at its start is bound when it appears
// (novox/hq issue 1790; mesh-sdk process.AwaitBucket).
sub = append(sub, stateCreatedHeard(kept)...)
// And its object buckets (novox/hq ADR 0592 §5, §9): its own written, others' its owners grant it read only.
pub = append(pub, objectGrants(p.Objects, p.ReadsObjects)...)
// And its logs (novox/hq ADR 0297): appended to and read directly, its own alone.
pub = append(pub, logGrants(p.Module, p.Logs)...)
pub = append(pub, readLogGrants(p.ReadLogs)...)
@@ -1026,6 +1031,7 @@ func PermissionsFor(p Principal) (Permissions, error) {
for _, d := range p.Carries {
pub = append(pub, stateGrants(stateAccess{Module: d.Module, Node: p.Node, Keeps: stateNames(d.State),
PerMachine: perMachineNames(d.State), Reads: d.Reads, KeyedReads: d.KeyedReads, Nodes: p.Nodes})...)
pub = append(pub, objectGrants(objectNames(d.Objects), d.ReadsObjects)...)
// A bundle reads every node's compositions on the runtime's own connection (ADR 0593 §6), which hears a node's
// bucket made later by any stream's creation.
if slices.Contains(d.Reads, delivery.ComposedRead) && !slices.Contains(sub, familyCreatedAdvisory) {
@@ -0,0 +1,116 @@
package buscompose
import (
"fmt"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// A module's own object bucket (novox/hq ADR 0592 §5, §9). The grants part creates it once, create-only, kept with no
// age limit and refusing a new object when full; only its module writes it; another module reads it only through a
// `reads` entry naming it and the owner's trusted config value of the bucket's grant naming that module, read-only, as a
// restricted stream is heard (ADR 0363 §4, §5). The condition when it passes 75 percent of its cap is the grants part's
// (grants.FullCondition).
// ObjectBucket is one object bucket a module owns, as the bus holds it.
type ObjectBucket struct {
Module string
Name string
// MaxMiB is its cap, catalogue.ObjectsLeastMiB to catalogue.ObjectsMostMiB.
MaxMiB int
// Restricted says it names a grant, so a module the owner's trusted config value names reads it.
Restricted bool
}
// Bucket is its name on the bus, `<module>_<name>` (catalogue.ObjectsBucket).
func (o ObjectBucket) Bucket() string { return catalogue.ObjectsBucket(o.Module, o.Name) }
// Stream is the stream the bus keeps it in.
func (o ObjectBucket) Stream() string { return "OBJ_" + o.Bucket() }
// ObjectsStreamConfig is a module's object bucket as the stream the client library's object store makes of it, with
// the mesh's rule: no age limit, its own cap, and a new object refused when full rather than an old one dropped (ADR
// 0592 §5); file storage and one replica, as a state's.
func ObjectsStreamConfig(o ObjectBucket) jetstream.StreamConfig {
b := o.Bucket()
return jetstream.StreamConfig{
Name: o.Stream(),
Description: fmt.Sprintf("%s's object bucket %q (novox/hq ADR 0592 §5): written by %s alone, read by the modules its "+
"grant names, kept with no age limit and at most %d MiB, a new object refused when full", o.Module, o.Name, o.Module, o.MaxMiB),
Subjects: []string{"$O." + b + ".C.>", "$O." + b + ".M.>"},
MaxBytes: int64(o.MaxMiB) * 1024 * 1024,
Storage: jetstream.FileStorage,
Replicas: 1,
Discard: jetstream.DiscardNew,
AllowRollup: true,
AllowDirect: true,
MaxMsgs: -1,
MaxConsumers: -1,
}
}
// objectReadGrants is what reading one object bucket needs and nothing more: binding it, a direct get of an object's
// metadata, an ordered consumer on its chunks or its names (a list or a watch), that consumer's deletion and flow
// control. No publish under `$O.`, so no write; no purge.
func objectReadGrants(bucket string) []string {
stream := "OBJ_" + bucket
return []string{
"$JS.API.STREAM.INFO." + stream,
"$JS.API.DIRECT.GET." + stream + ".>",
"$JS.API.CONSUMER.CREATE." + stream + ".>",
"$JS.API.CONSUMER.DELETE." + stream + ".>",
"$JS.FC." + stream + ".>",
}
}
// objectGrants is what a principal publishes for its module's own object buckets (written, and a replaced or deleted
// object's chunks purged) and for those it reads (read-only), never their creation, which is the grants part's.
func objectGrants(own, reads []string) []string {
var out []string
for _, b := range own {
if !safeSubject.MatchString(b) {
continue
}
out = append(out, "$O."+b+".>", "$JS.API.STREAM.PURGE.OBJ_"+b)
out = append(out, objectReadGrants(b)...)
}
for _, b := range reads {
if safeSubject.MatchString(b) {
out = append(out, objectReadGrants(b)...)
}
}
return out
}
// objectNames is the bus names of a module's own object buckets.
func objectNames(objects []ObjectBucket) []string {
out := make([]string, 0, len(objects))
for _, o := range objects {
out = append(out, o.Bucket())
}
return out
}
// writesOwnObjects is a module writing one of its own object buckets (ADR 0592 §5), itself or through the runtime that
// carries it, and nothing of another's.
func writesOwnObjects(p Principal, pattern string) bool {
switch p.Kind {
case KindModule:
for _, b := range p.Objects {
if pattern == "$O."+b+".>" {
return true
}
}
case KindNodeTools:
for _, d := range p.Carries {
for _, b := range objectNames(d.Objects) {
if pattern == "$O."+b+".>" {
return true
}
}
}
}
return false
}
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -25,10 +25,14 @@ import (
// Declared is one module on one node, as composing its authority needs it.
type Declared struct {
Module string
Emits []string
Consumes []string
Serves []string
Module string
// Objects are its own object buckets, and ReadsObjects the bus names of the object buckets of others its reads name
// and their owners' grants let it read (novox/hq ADR 0592 §5, §9).
Objects []ObjectBucket
ReadsObjects []string
Emits []string
Consumes []string
Serves []string
// Served are the config keys its served values and zone name fill from its settings (catalogue.ServedSettings): what
// every node may read of its resolved values where it runs (Principal.ServedReads, issue 1126).
Served []string
@@ -244,6 +248,7 @@ func Users(r Records) ([]Principal, error) {
State: stateNames(d.State), Reads: d.Reads, ReadLogs: d.ReadLogs, SnapshotsTheBus: d.SnapshotsTheBus,
PerMachine: perMachineNames(d.State), KeyedReads: d.KeyedReads,
Logs: logNames(d.Logs), Streams: d.Streams, Hears: ComposedHears(d), Queues: d.Queues,
Objects: objectNames(d.Objects), ReadsObjects: d.ReadsObjects,
})
// **The grants part creates what a manifest declares** (novox/hq ADR 0547 §5; the operator, 2026-10-26
// 08:54): every placed module's buckets, logs and streams, each by its stream's name.
@@ -539,6 +544,9 @@ func declaredStreamsOf(r Records) []string {
for _, s := range d.Streams {
seen[s.Name] = true
}
for _, o := range d.Objects {
seen[o.Stream()] = true
}
}
}
out := make([]string, 0, len(seen))
@@ -224,9 +224,10 @@ var WritersTable = []WriterRow{
"node's latest declaration no longer names it",
Others: "a node-engine reads an object of its own bucket by its name and may neither list the bucket nor write it",
Subjects: []string{"$O.*.>"}, Writes: func(p Principal, pattern string) bool {
return grantsOwnObjects(p, pattern)
return grantsOwnObjects(p, pattern) || writesOwnObjects(p, pattern)
},
Shared: "nobody else: the grants part writes its own object bucket " + GrantsUserListObjects + ", the user list sealed to the bus node, and no other (novox/hq ADR 0547 clause 7)"},
Shared: "nobody else: the grants part writes its own object bucket " + GrantsUserListObjects + ", the user list sealed to the bus node, and no other (novox/hq ADR 0547 clause 7); " +
"a module writes the object buckets its manifest declares, each its own alone (novox/hq ADR 0592 §5)"},
// What a machine's provisioner client placed of a module's files (novox/hq ADR 0349 §4, design 57 P6): its own
// machine's word alone, which a walk's end and a gate wait for (ADR 0362, ADR 0349 §15). Kept newest per module
// in the stream ProvAppliedStream, and by the client in the provisioner server's `applied` bucket.
@@ -88,6 +88,10 @@ func Configs(m catalogue.Manifest, nodes []string) map[string]jetstream.StreamCo
c := buscompose.ModuleStreamConfig(s)
out[c.Name] = c
}
for _, o := range objectsOf(m) {
c := buscompose.ObjectsStreamConfig(o)
out[c.Name] = c
}
return out
}
@@ -184,14 +188,25 @@ func Composed(m catalogue.Manifest, nodes []string) []Item {
out = append(out, Item{Kind: "stream", Stream: st.Name, Composed: Limits{Subjects: set(st.Subjects),
Retention: string(st.Retention), MaxAgeHours: st.MaxAge / 3600, MaxBytes: st.MaxBytes}})
}
for _, o := range objectsOf(m) {
c := buscompose.ObjectsStreamConfig(o)
out = append(out, Item{Kind: "objects", Stream: c.Name, Composed: Limits{Subjects: set(c.Subjects),
Retention: string(buscompose.RetentionLimits), MaxBytes: c.MaxBytes}})
}
return out
}
// HeldStreams reads every stream the bus holds through its module's nats_streams, by name.
func HeldStreams(c Caller) (map[string]Limits, error) {
held, _, err := HeldStreamsAndBytes(c)
return held, err
}
// HeldStreamsAndBytes is HeldStreams with each stream's size in bytes, by name (for Full).
func HeldStreamsAndBytes(c Caller) (map[string]Limits, map[string]int64, error) {
raw, err := c.Call(CallStreams, map[string]any{})
if err != nil {
return nil, fmt.Errorf("%s: %w", CallStreams, err)
return nil, nil, fmt.Errorf("%s: %w", CallStreams, err)
}
var got struct {
Detail []struct {
@@ -200,16 +215,18 @@ func HeldStreams(c Caller) (map[string]Limits, error) {
Retention string `json:"retention"`
MaxAgeHours int `json:"max_age_hours"`
MaxPerSubject int `json:"max_per_subject"`
Bytes int64 `json:"bytes"`
} `json:"detail"`
}
if err := json.Unmarshal(unwrap(raw), &got); err != nil {
return nil, fmt.Errorf("%s answered no streams: %w", CallStreams, err)
return nil, nil, fmt.Errorf("%s answered no streams: %w", CallStreams, err)
}
out := map[string]Limits{}
out, bytes := map[string]Limits{}, map[string]int64{}
for _, s := range got.Detail {
out[s.Name] = Limits{Subjects: set(s.Subjects), Retention: s.Retention, MaxAgeHours: s.MaxAgeHours, MaxPerSubject: s.MaxPerSubject}
bytes[s.Name] = s.Bytes
}
return out, nil
return out, bytes, nil
}
// Assert composes a module's items and compares each with what the bus holds; held nil is not read. It asserts nothing.
@@ -0,0 +1,95 @@
package grants
import (
"fmt"
"strings"
"git.novox.be/novox/mesh-sdk/go/asks"
"git.novox.be/novox/mesh-sdk/go/catalogue"
)
// **A history stream or an object bucket past 75 percent of its cap is said, in one place for every module** (novox/hq
// ADR 0592 §6, §25; ADR 0593 §10 gives the controller's part to this part). Such a stream or bucket refuses a new message
// or object when full rather than dropping its oldest, so filling up is the one way it fails: the part reads every
// stream's size from the bus module's nats_streams at each pass (HeldStreams) and raises the warning FullCondition naming
// each one past catalogue.FullWarnPercent, cleared once none is.
// FullCondition is the key of the warning raised while a history stream or an object bucket is past its mark.
const FullCondition = "mesh-grants.history-full"
// Fill is how full one history stream or object bucket is: its module, its stream on the bus, its bytes and its cap.
type Fill struct {
Module string `json:"module"`
Stream string `json:"stream"`
Bytes int64 `json:"bytes"`
Cap int64 `json:"cap"`
}
// Percent is its bytes in percent of its cap, rounded down.
func (f Fill) Percent() int64 { return f.Bytes * 100 / f.Cap }
// Full is every history stream and object bucket of a placed module past catalogue.FullWarnPercent of its cap, by the
// sizes the bus holds (bytes, by stream name), in module and stream order. One the bus does not hold is not full. Pure.
func Full(in Inputs, bytes map[string]int64) []Fill {
var out []Fill
for _, module := range sortedKeys(in.Manifests) {
if !assigned(in, module) {
continue
}
m := in.Manifests[module]
var capped []Fill
for _, s := range m.Streams {
if s.KeepsHistory() {
capped = append(capped, Fill{Module: module, Stream: s.BusName(), Cap: int64(s.Cap()) << 20})
}
}
for _, o := range objectsOf(m) {
capped = append(capped, Fill{Module: module, Stream: "OBJ_" + o.Bucket(), Cap: int64(o.MaxMiB) << 20})
}
for _, f := range capped {
held, ok := bytes[f.Stream]
if !ok || f.Cap <= 0 {
continue
}
f.Bytes = held
if f.Bytes*100 >= catalogue.FullWarnPercent*f.Cap {
out = append(out, f)
}
}
}
return out
}
// sayFull raises FullCondition while a history stream or an object bucket is past its mark, and clears it once none is:
// said once per set of streams, again when that set changes, and at the first pass after a start.
func (p *Part) sayFull(full []Fill) {
if p.Say == nil {
return
}
var names, lines []string
for _, f := range full {
names = append(names, f.Stream)
lines = append(lines, fmt.Sprintf("%s of %s: %d of %d MiB (%d%%)", f.Stream, f.Module, f.Bytes>>20, f.Cap>>20, f.Percent()))
}
said := strings.Join(names, " ")
p.holdMu.Lock()
defer p.holdMu.Unlock()
if p.fullKnown && said == p.fullSaid {
return
}
c := asks.Condition{Key: FullCondition, Subject: asks.Scope{Scope: "module", ID: Module}}
if said == "" {
c.State = asks.StateCleared
} else {
c.Severity, c.State = asks.SeverityWarning, asks.StateRaised
c.Summary = fmt.Sprintf("%d history stream(s) or object bucket(s) past %d%% of their cap: %s", len(full),
catalogue.FullWarnPercent, strings.Join(lines, "; "))
c.Headline, c.Needs = "A stream that keeps its history is filling up", "a larger cap in the module's manifest, or data removed"
c.Explanation = asks.NeedsYou + "a decision before it is full: such a stream or bucket refuses new messages when full " +
"rather than dropping its oldest (novox/hq ADR 0592 §6). " + strings.Join(lines, "; ")
c.Resolved = "every history stream and object bucket is below its mark"
}
if p.Say(c) == nil {
p.fullSaid, p.fullKnown = said, true
}
}
+10 -1
View File
@@ -96,6 +96,8 @@ func (s UserListState) inputs() string {
// Result is what one composition did.
type Result struct {
// Full is every history stream and object bucket past its mark at this pass (full.go, ADR 0592 §6).
Full []Fill `json:"full,omitempty"`
UserList UserListState `json:"user_list"`
Revision uint64 `json:"revision"`
Written bool `json:"written"`
@@ -181,6 +183,9 @@ type Part struct {
behindSaid string
behindKnown bool
behindSeen map[string]bool
// fullSaid is the history streams and object buckets past their mark last said as a condition, once fullKnown (full.go).
fullSaid string
fullKnown bool
// refusedSaid is the refused principals last said as a condition, once refusedKnown (refused.go).
refusedSaid string
refusedKnown bool
@@ -359,7 +364,11 @@ func (p *Part) Compose() Result {
res.Retired[key] = m.Retired
}
}
held, heldErr := HeldStreams(p.Call)
held, sizes, heldErr := HeldStreamsAndBytes(p.Call)
if heldErr == nil {
res.Full = Full(in, sizes)
p.sayFull(res.Full)
}
var ask Ask
if p.Create != nil || p.Consumers != nil {
ask = p.asked()
+38 -1
View File
@@ -80,6 +80,7 @@ func records(in Inputs, union bool) (buscompose.Records, []Gap) {
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)
d.Reads, d.ReadsObjects, d.ReadsRefused = readsOfObjects(m.Module, node, d.Reads, d.ReadsRefused, in.Manifests, values)
return d
}
d := declare(m)
@@ -136,6 +137,8 @@ func declaredFor(m catalogue.Manifest, seats map[string]catalogue.SeatDeclaratio
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),
// Its own object buckets (ADR 0592 §5).
Objects: objectsOf(m),
}
d.Loads = loadsABundle(m)
if _, reads := m.OwnSecrets["broker"]; !reads {
@@ -245,6 +248,40 @@ func bucketsFor(m catalogue.Manifest, nodes []string) []buscompose.Bucket {
return append(out, buscompose.EngineComposedBuckets(nodes)...)
}
// objectsOf is a manifest's own object buckets (ADR 0592 §5).
func objectsOf(m catalogue.Manifest) []buscompose.ObjectBucket {
var out []buscompose.ObjectBucket
for _, o := range m.Objects {
out = append(out, buscompose.ObjectBucket{Module: m.Module, Name: o.Name, MaxMiB: o.MaxMiB, Restricted: o.Grant != nil})
}
return out
}
// readsOfObjects takes out of a module's reads each that names another module's object bucket (ADR 0592 §9): read, by
// its bus name, where the owner's trusted config value of the bucket's grant names the module on this node, and refused
// otherwise, a bucket with no grant included (its owner alone reads it). Every other read stays as it was: a state's.
func readsOfObjects(module, node string, reads, refused []string, manifests map[string]catalogue.Manifest,
values func(provider, setting string) []any) (states, objects, refusedOut []string) {
refusedOut = refused
for _, r := range reads {
owner, name, ok := strings.Cut(strings.TrimSpace(r), ".")
o, isObjects := catalogue.ObjectsDeclaration{}, false
if ok {
o, isObjects = manifests[owner].ObjectsNamed(name)
}
if !isObjects {
states = append(states, r)
continue
}
if o.Grant != nil && o.ObjectsReader(values(owner, o.Grant.Config), module, node) {
objects = append(objects, o.Bucket(owner))
} else {
refusedOut = append(refusedOut, r)
}
}
return states, objects, refusedOut
}
func logsOf(m catalogue.Manifest) []buscompose.Log {
var out []buscompose.Log
for _, l := range m.Logs {
@@ -276,7 +313,7 @@ func streamsOf(m catalogue.Manifest) []buscompose.ModuleStream {
}
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})
WorkQueue: s.Retention == catalogue.StreamRetentionWorkQueue, Duplicates: s.Duplicates, History: s.KeepsHistory()})
}
return out
}