W8.7 (kept events): the grants part makes each kept seat event's stream, sourced from the events stream, and each consuming module's durable on it from the last message per subject

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

Builds: design 47 G §260, §331, §396
This commit is contained in:
2026-11-11 14:39:23 +01:00
parent c4a8590671
commit 28ebd45863
12 changed files with 330 additions and 5 deletions
@@ -7,6 +7,7 @@ import (
stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"git.novox.be/novox/mesh-sdk/go/seatlist"
)
@@ -31,6 +32,14 @@ func (g *seatGate) MayAct(seats ...string) bool {
return r != nil && r.MayAct(seats...)
}
// Definitions are the seat definitions the reader holds: what the part makes each kept event's stream from.
func (g *seatGate) Definitions() map[string]seatdef.Definition {
if r := g.reader.Load(); r != nil {
return r.Definitions()
}
return nil
}
// follow starts the reader over bus: what it holds is given to catalogue at once and at every change, and changed is called
// then (a pass composes again).
func (g *seatGate) follow(bus seatlist.Bus, changed func(), logf func(string, ...any)) *seatlist.Reader {
+28
View File
@@ -8,6 +8,7 @@ import (
"errors"
"fmt"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"io/fs"
"maps"
"os"
@@ -223,6 +224,14 @@ func (b *fakeBus) CreateDeclared(cfg jetstream.StreamConfig) error {
return nil
}
// keptGate is a seat list that has been read and holds one definition.
type keptGate struct{ def seatdef.Definition }
func (keptGate) MayAct(...string) bool { return true }
func (g keptGate) Definitions() map[string]seatdef.Definition {
return map[string]seatdef.Definition{g.def.Seat: g.def}
}
func newFakeBus() *fakeBus {
return &fakeBus{states: map[string]map[string][]byte{}, refused: map[string]bool{}, own: map[string]map[string][]byte{},
revs: map[string]uint64{}, refuseTo: map[string]bool{}}
@@ -2060,6 +2069,25 @@ func TestGrants(t *testing.T) {
t.Error("the bucket the bus holds was asked to be created")
}
}
// A seat's event the seat list says is kept in a stream of its own (design 47): its stream is made once, and the module that
// consumes it has its durable on that stream, none on the shared one.
kb := fixture()
kb.manifestOf("mod-y", `{"module":"mod-y","version":"1","own-secrets":{"broker":"/var/lib/y/broker"},"consumes":["mesh-conditions.condition-raised"]}`)
kb.loaded = shadowOf(t, kb)
kp := partOn(kb)
kp.Create, kp.Consumers = kb, kb
kp.SeatList = keptGate{seatdef.Definition{Seat: "mesh-conditions", Events: []seatdef.Message{
{Name: "condition-raised", Keep: seatdef.KeepStream, MaxAgeDays: 7, MaxMiB: 64}}}}
kp.Compose()
if !slices.Contains(kb.creates, "SEAT_mesh-conditions_condition-raised") || !slices.Contains(kb.consumerCreates, "SEAT_mesh-conditions_condition-raised.b_mod-y") ||
strings.Contains(fmt.Sprint(kb.consumers["EVENTS"]), "condition-raised") ||
!strings.Contains(fmt.Sprint(kb.consumers["SEAT_mesh-conditions_condition-raised"]), "mesh.seat.mesh-conditions.event.condition-raised") {
t.Errorf("a kept event stream %v and durable %v, now %v", kb.creates, kb.consumerCreates, kb.consumers)
}
before := len(kb.creates)
if kp.Compose(); len(kb.creates) != before {
t.Errorf("a kept stream the bus holds was made again: %v", kb.creates)
}
})
// 774 (ADR 0563 §2, §3): the node-engine's manifest (mesh-host) is placed on every node though no node is assigned it,
@@ -111,5 +111,8 @@ func ConsumerConfig(c Consumer) jetstream.ConsumerConfig {
if c.FromNow {
cfg.DeliverPolicy = jetstream.DeliverNewPolicy
}
if c.LastPerSubject {
cfg.DeliverPolicy = jetstream.DeliverLastPerSubjectPolicy
}
return cfg
}
@@ -1,15 +1,28 @@
package buscompose
import (
"fmt"
"testing"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// Defends issue 1790 (ADR 0547 §5): the grants part creates a declared bucket and stream as the controller asserts them.
func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) {
plain := BucketStreamConfig(Bucket{Module: "mesh-vault", Name: "bus-accounts"})
defs := map[string]seatdef.Definition{
"node-engine": {Seat: "node-engine", Events: []seatdef.Message{
{Name: "running", Subject: "running.<node>.<module>", Keep: seatdef.KeepLastPerSubject}, {Name: "composed"}}},
"mesh-delivery": {Seat: "mesh-delivery", Events: []seatdef.Message{
{Name: "owed", Subject: "owed.<repository>.<number>", Keep: seatdef.KeepStream, MaxAgeDays: 7, MaxMiB: 64}}},
}
kept := KeptEventsOf(defs)
running, owed := kept[1].StreamConfig(), kept[0].StreamConfig()
rest, heard := SplitKept(kept, []string{"mesh-catalog.manifest-accepted", "node-engine.running.*.*", "node-engine.running"})
heardConsumerCfg := ConsumerConfig(heardConsumer(Principal{Kind: KindModule, Node: "novox", Module: "mesh-grants"}, heard[0]))
keyed := BucketStreamConfig(Bucket{Module: "m", Name: "presence", History: 5, TTLSeconds: 60, KeyTTL: true})
queue := ModuleStreamConfig(ModuleStream{Module: "b", Name: "ASKS", Events: []string{"ask"}, MaxAgeDays: 7, MaxMiB: 32,
WorkQueue: true, Duplicates: 1})
@@ -38,6 +51,21 @@ func declaredStreamsAreCreatedAsTheControllerAssertsThem(t *testing.T) {
{"queue discards new", queue.Discard, jetstream.DiscardNew},
{"queue duplicates", queue.Duplicates, time.Hour},
{"queue metadata", queue.Metadata[WorkQueueMetadata] + queue.Metadata[RestrictedMetadata], "truetrue"},
{"kept events are two", len(kept), 2},
{"kept stream name", running.Name, "SEAT_node-engine_running"},
{"a kept stream takes no subject of the shared stream", len(running.Subjects), 0},
{"a kept stream sources the shared stream by the event's subject", running.Sources[0].Name + " " + running.Sources[0].FilterSubject,
"EVENTS mesh.seat.node-engine.event.running.*.*"},
{"last per subject holds one message a subject", running.MaxMsgsPerSubject, int64(1)},
{"last per subject has no age", running.MaxAge, time.Duration(0)},
{"the stream kept for a week", owed.MaxAge, 7 * 24 * time.Hour},
{"the stream kept within its size", owed.MaxBytes, int64(64 << 20)},
{"its name is the bus's", owed.Name, "SEAT_mesh-delivery_owed"},
{"a kept consume leaves the shared stream's", len(rest), 1},
{"both namings of one event are one durable on its stream", heard[0].Stream + " " + fmt.Sprint(heard[0].Filters),
"SEAT_node-engine_running [mesh.seat.node-engine.event.running.*.*]"},
{"its durable starts at the last message of each subject", heardConsumerCfg.DeliverPolicy, jetstream.DeliverLastPerSubjectPolicy},
{"its durable is the module's on the node", heardConsumerCfg.Durable, "novox_mesh-grants"},
} {
if row.got != row.want {
t.Errorf("%s: %v, want %v", row.what, row.got, row.want)
@@ -2,6 +2,7 @@ package buscompose
import (
"fmt"
"slices"
"sort"
"strings"
)
@@ -60,7 +61,9 @@ type Consumer struct {
// holds, replays every merge and every build of that week as if it had just happened. A consumer
// that exists keeps where it is, whatever this says; only its making is decided here.
FromNow bool
Why string
// LastPerSubject makes the consumer start from the last message of each subject (a kept event, kept.go).
LastPerSubject bool
Why string
// Resettable says why this consumer may be re-made to deliver from now by the mesh itself, with
// nobody asked (novox/hq to-be 45 §7, healer H4): what a reset drops, something else catches up.
// Empty for every consumer where nothing would — a module's, whose events would be lost to it.
@@ -155,7 +158,8 @@ func ConsumerFor(p Principal) (Consumer, bool) {
var filters []string
for _, s := range perms.Subscribe {
// Not a module stream's, which its own consumer on that stream hears (novox/hq ADR 0363).
if strings.Contains(s, ".event.") && !strings.HasPrefix(s, moduleStreamPrefix) {
// Nor a kept seat event's, which the module's durable on its own stream hears (kept.go).
if strings.Contains(s, ".event.") && !strings.HasPrefix(s, moduleStreamPrefix) && !heardInKept(p.Hears, s) {
filters = append(filters, s)
}
}
@@ -385,3 +389,14 @@ func upperSnake(s string) string {
}
return string(out)
}
// heardInKept says the subject is one a hearing of a kept seat event's stream filters (kept.go): the module's consumer on
// that stream carries it, never its consumer on EVENTS.
func heardInKept(hears []Heard, subject string) bool {
for _, h := range hears {
if strings.HasPrefix(h.Stream, "SEAT_") && slices.Contains(h.Filters, subject) {
return true
}
}
return false
}
@@ -0,0 +1,127 @@
package buscompose
import (
"slices"
"sort"
"strings"
"time"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/seatdef"
)
// **A seat's event kept in a stream of its own** (novox/hq design 47, *What each step says*; the seat definition's `keep`):
// the node-engine's `running.<node>.<module>` last per subject with no age limit, the gate's and the checker's `owed` for a
// week within 64 MiB. The event's subject lies under `mesh.seat.*.event.>`, which the shared events stream already takes, and
// the bus refuses two streams taking one subject: so the kept stream takes no subject of its own but sources the shared
// stream's messages of that subject, and keeps them by its own limits (the shared stream's seven days never end a
// `running`). The part makes the stream, and each hearing module's durable on it from the module's `consumes` (the bus name
// is seatdef.KeptStream's, the one name for every part that hears it).
// KeptEvent is a seat's event with a stream of its own.
type KeptEvent struct {
Seat string
Event string
// Stream is its name on the bus (seatdef.KeptStream).
Stream string
// Filter is the subject that carries every instance of the event, a part of its subject `*`.
Filter string
// Keep is seatdef.KeepLastPerSubject or seatdef.KeepStream; MaxAgeDays and MaxMiB bound the latter.
Keep string
MaxAgeDays int
MaxMiB int
Why string
}
// KeptEventsOf is every event the seat definitions keep in a stream of its own, sorted by stream name. Pure.
func KeptEventsOf(defs map[string]seatdef.Definition) []KeptEvent {
var out []KeptEvent
for _, d := range defs {
for _, e := range d.Events {
if e.Keep == "" {
continue
}
out = append(out, KeptEvent{Seat: d.Seat, Event: e.Name, Stream: seatdef.KeptStream(d.Seat, e),
Filter: seatdef.EventSubject(d.Seat, e.Local()), Keep: e.Keep, MaxAgeDays: e.MaxAgeDays, MaxMiB: e.MaxMiB,
Why: "the seat " + d.Seat + "'s event " + e.Name + ", kept apart from the shared events stream (design 47): " + keptHow(e)})
}
}
sort.Slice(out, func(i, j int) bool { return out[i].Stream < out[j].Stream })
return out
}
func keptHow(e seatdef.Message) string {
if e.Keep == seatdef.KeepLastPerSubject {
return "the last message of each subject, no age limit"
}
return "every message within its age and size"
}
// LastPerSubject says a hearing consumer starts from the last message of each subject, never from now.
func (k KeptEvent) LastPerSubject() bool { return k.Keep == seatdef.KeepLastPerSubject }
// StreamConfig is the kept stream as it is created: sourced from the shared events stream by the event's subject, with
// its own limits. Last per subject holds one message a subject and no age; the other form its age and size, a full stream
// refusing rather than dropping what it holds.
func (k KeptEvent) StreamConfig() jetstream.StreamConfig {
cfg := jetstream.StreamConfig{
Name: k.Stream,
Description: k.Why,
Storage: jetstream.FileStorage,
Replicas: 1,
Sources: []*jetstream.StreamSource{{Name: EventsStream, FilterSubject: k.Filter}},
}
if k.LastPerSubject() {
cfg.MaxMsgsPerSubject = 1
return cfg
}
cfg.MaxAge = time.Duration(k.MaxAgeDays) * 24 * time.Hour
cfg.MaxBytes = int64(k.MaxMiB) << 20
cfg.Discard = jetstream.DiscardNew
return cfg
}
// KeptHeard is the part of a module's `consumes` (`<seat>.<event>`) that names a kept event, as the Heard it makes: the
// module's durable on the kept stream, filtered to the subject it consumes, from the last message of each subject for an
// event so kept. ok is false for a consume that names none.
func KeptHeard(kept []KeptEvent, consume string) (h Heard, ok bool) {
seat, rest, named := strings.Cut(consume, ".")
if !named {
return Heard{}, false
}
for _, k := range kept {
local := strings.TrimPrefix(k.Filter, seatdef.EventSubject(k.Seat, ""))
if k.Seat != seat || (rest != k.Event && rest != local) {
continue
}
return Heard{Stream: k.Stream, Emitter: k.Seat, Filters: []string{k.Filter}, LastPerSubject: k.LastPerSubject()}, true
}
return Heard{}, false
}
// SplitKept divides a module's consumes into the ones the shared events stream carries and the kept events' hearing, merged
// by stream, filters sorted without repeats.
func SplitKept(kept []KeptEvent, consumes []string) (rest []string, hears []Heard) {
byStream := map[string]*Heard{}
var order []string
for _, c := range consumes {
h, ok := KeptHeard(kept, c)
if !ok {
rest = append(rest, c)
continue
}
if byStream[h.Stream] == nil {
byStream[h.Stream] = &Heard{Stream: h.Stream, Emitter: h.Emitter, LastPerSubject: h.LastPerSubject}
order = append(order, h.Stream)
}
byStream[h.Stream].Filters = append(byStream[h.Stream].Filters, h.Filters...)
}
for _, s := range order {
h := byStream[s]
slices.Sort(h.Filters)
h.Filters = slices.Compact(h.Filters)
hears = append(hears, *h)
}
return rest, hears
}
@@ -169,6 +169,9 @@ type Heard struct {
// Restricted says the stream is restricted by its provider's trusted config value (ADR 0363 §4): then it
// is heard only by a module on an account of its own (ADR 0376 §1, ComposedHears).
Restricted bool
// LastPerSubject says the stream is a seat's event kept last per subject (kept.go): the module's consumer starts from the
// last message of each subject, never from now.
LastPerSubject bool
}
// ComposedHears is what of a module's hearing the controller composes (novox/hq ADR 0376 §1): every stream it
@@ -235,8 +238,9 @@ func heardConsumer(p Principal, h Heard) Consumer {
Filters: filters,
AckWaitSeconds: 30,
MaxDeliver: 5,
// From now, like a module's consumer on EVENTS (novox/hq ADR 0443 §1).
FromNow: true,
// From now, like a module's consumer on EVENTS (novox/hq ADR 0443 §1); a kept event's from its last message per subject.
FromNow: !h.LastPerSubject,
LastPerSubject: h.LastPerSubject,
Why: "what " + p.Module + " consumes of " + h.Emitter + "'s stream " + h.Stream + " and is granted " +
"(novox/hq ADR 0363); after max-deliver it gives an event up, and the controller keeps it in " +
"DEAD_LETTERS until a person delivers it again or drops it",
@@ -146,6 +146,9 @@ type ProcessPrincipal struct {
// Records is what composing a user list needs to know about the mesh, and nothing more.
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
// 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.
@@ -577,6 +580,9 @@ func runnerServes(d Declared) bool {
// name. What the grants part may create, and nothing beside.
func declaredStreamsOf(r Records) []string {
seen := map[string]bool{}
for _, k := range r.Kept {
seen[k.Stream] = true
}
placed := [][]Declared{}
for _, mods := range r.Assigned {
placed = append(placed, mods)
@@ -177,6 +177,9 @@ type Inputs struct {
// Refused is every manifest the indexer holds that the part cannot read or mesh-sdk's reader refuses, assigned or
// not, each with why: its module counts as placed nowhere, so it is said loudly, never passed over (issue 1945).
Refused []Gap
// 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
// 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
@@ -340,6 +343,9 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
NoFingerprint: map[string]string{}, Running: map[string]catalogue.Manifest{}, Had: map[string]catalogue.Manifest{}, HadFailed: map[string]string{},
Values: map[string]any{}, Gaps: append([]Gap(nil), standingGaps...)}
in.SeatListUnread = gate != nil && !gate.MayAct()
if d, ok := gate.(SeatDefinitions); ok && !in.SeatListUnread {
in.Kept = buscompose.KeptEventsOf(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
// assignments are read no more, and the seats held once for the mesh are held where it says. Not read is a failed
@@ -0,0 +1,86 @@
package grants
import (
"errors"
"fmt"
"slices"
"git.novox.be/novox/mesh-sdk/go/seatdef"
"mesh-grants/internal/buscompose"
)
// **A seat's kept event is heard on its own stream** (novox/hq design 47; buscompose/kept.go): a module that consumes one has
// its durable on the kept stream, from the last message of each subject for an event kept so, and no filter of it on the
// shared events stream.
// SeatDefinitions is what a seat gate may also say: the definitions the seat list holds (seatlist.Reader.Definitions).
type SeatDefinitions interface {
Definitions() map[string]seatdef.Definition
}
// keptOutOfHearing moves each seat event in watches or in the module's consumes that a kept stream carries out of the shared
// stream's hearing and into hears, one durable on the kept stream each (merged by stream). A watch left with no event is
// dropped. It answers what is left to watch and consume, the hears, and whether any moved.
func keptOutOfHearing(kept []buscompose.KeptEvent, watches []buscompose.Seat, consumes []string, hears []buscompose.Heard) ([]buscompose.Seat, []string, []buscompose.Heard, bool) {
if len(kept) == 0 {
return watches, consumes, hears, false
}
var left []buscompose.Seat
var named []string // every consume and every watched event, as `<seat>.<event>`
named = append(named, consumes...)
for _, w := range watches {
var emits []string
for _, e := range w.Emits {
if _, ok := buscompose.KeptHeard(kept, w.Name+"."+e); ok {
named = append(named, w.Name+"."+e)
continue
}
emits = append(emits, e)
}
if len(emits) > 0 {
w.Emits = emits
left = append(left, w)
}
}
rest, moved := buscompose.SplitKept(kept, consumes)
_, movedWatched := buscompose.SplitKept(kept, named[len(consumes):])
moved = append(moved, movedWatched...)
if len(moved) == 0 {
return watches, consumes, hears, false
}
for _, h := range moved {
if i := slices.IndexFunc(hears, func(x buscompose.Heard) bool { return x.Stream == h.Stream }); i >= 0 {
hears[i].Filters = slices.Compact(slices.Sorted(slices.Values(append(slices.Clone(hears[i].Filters), h.Filters...))))
continue
}
hears = append(hears, h)
}
return left, rest, hears, true
}
// createKept makes each kept event's stream the bus lacks, from the shared events stream's messages of its subject, and
// says what it made. A stream the bus holds is left as it is; one made at this seat list and gone since is made again only on
// a hand assert (the same rule as a manifest's streams: made once per definition). held is what the bus held at this pass.
func createKept(c Creator, done *creations, kept []buscompose.KeptEvent, held map[string]Limits, force bool) (made, notMade []string) {
for _, k := range kept {
if _, ok := held[k.Stream]; ok {
continue
}
cfg := k.StreamConfig()
digest := fmt.Sprintf("%v", cfg.Sources[0].FilterSubject) + k.Keep + fmt.Sprint(k.MaxAgeDays, k.MaxMiB)
if !force && done.madeAt(k.Stream) == digest {
notMade = append(notMade, k.Stream+": made at this definition and gone since; grants_assert makes it again")
continue
}
switch err := c.CreateDeclared(cfg); {
case err == nil:
done.keep(k.Stream, digest)
made = append(made, k.Stream)
case errors.Is(err, ErrHeld):
default:
notMade = append(notMade, k.Stream+": "+err.Error()+" (tried again at the next event)")
}
}
return made, notMade
}
@@ -498,6 +498,16 @@ func (p *Part) Compose() Result {
ask = p.asked()
}
forced := ask.Assert
if p.Create != nil && heldErr == nil {
made, notMade := createKept(p.Create, &p.made, in.Kept, held, forced != "")
if len(made) > 0 {
res.Asserted = append(res.Asserted, "the seats' kept events: created "+strings.Join(made, ", "))
p.say(&res, EventAsserted, map[string]any{"module": Module, "created": made})
}
for _, why := range notMade {
res.NotAsserted = append(res.NotAsserted, "kept event "+why)
}
}
for _, module := range sortedKeys(in.Manifests) {
if !assigned(in, module) {
continue
@@ -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,
NodeLogins: in.NodeLogins, Kept: in.Kept,
// 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 {
@@ -95,6 +95,9 @@ 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, declared, node, values, d.Consumes)
var keptHeard bool
d.Watches, d.Consumes, d.Hears, keptHeard = keptOutOfHearing(in.Kept, d.Watches, d.Consumes, d.Hears)
d.HearsAStream = d.HearsAStream || keptHeard
d.Reads, d.ReadsObjects, d.ReadsRefused = readsOfObjects(m.Module, node, d.Reads, d.ReadsRefused, declared, values)
// 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).