diff --git a/modules/mesh-grants/cmd/mesh-grants/seatlist.go b/modules/mesh-grants/cmd/mesh-grants/seatlist.go index 8ad180bc..7b30bb59 100644 --- a/modules/mesh-grants/cmd/mesh-grants/seatlist.go +++ b/modules/mesh-grants/cmd/mesh-grants/seatlist.go @@ -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 { diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 74f7bacb..e4a02217 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -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, diff --git a/modules/mesh-grants/internal/buscompose/create.go b/modules/mesh-grants/internal/buscompose/create.go index f4d76a68..07d87f16 100644 --- a/modules/mesh-grants/internal/buscompose/create.go +++ b/modules/mesh-grants/internal/buscompose/create.go @@ -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 } diff --git a/modules/mesh-grants/internal/buscompose/create_test.go b/modules/mesh-grants/internal/buscompose/create_test.go index 0df9580e..7deb3d6f 100644 --- a/modules/mesh-grants/internal/buscompose/create_test.go +++ b/modules/mesh-grants/internal/buscompose/create_test.go @@ -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..", Keep: seatdef.KeepLastPerSubject}, {Name: "composed"}}}, + "mesh-delivery": {Seat: "mesh-delivery", Events: []seatdef.Message{ + {Name: "owed", Subject: "owed..", 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) diff --git a/modules/mesh-grants/internal/buscompose/derived.go b/modules/mesh-grants/internal/buscompose/derived.go index 4533d463..1bb8df36 100644 --- a/modules/mesh-grants/internal/buscompose/derived.go +++ b/modules/mesh-grants/internal/buscompose/derived.go @@ -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 +} diff --git a/modules/mesh-grants/internal/buscompose/kept.go b/modules/mesh-grants/internal/buscompose/kept.go new file mode 100644 index 00000000..77fbb31c --- /dev/null +++ b/modules/mesh-grants/internal/buscompose/kept.go @@ -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..` 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` (`.`) 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 +} diff --git a/modules/mesh-grants/internal/buscompose/modstreams.go b/modules/mesh-grants/internal/buscompose/modstreams.go index 9a0ceb83..64e0cec1 100644 --- a/modules/mesh-grants/internal/buscompose/modstreams.go +++ b/modules/mesh-grants/internal/buscompose/modstreams.go @@ -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", diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index 839339a7..e7f2a49f 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -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) diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index d885b67f..ae44400a 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -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 diff --git a/modules/mesh-grants/internal/grants/kept.go b/modules/mesh-grants/internal/grants/kept.go new file mode 100644 index 00000000..31e03251 --- /dev/null +++ b/modules/mesh-grants/internal/grants/kept.go @@ -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 `.` + 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 +} diff --git a/modules/mesh-grants/internal/grants/part.go b/modules/mesh-grants/internal/grants/part.go index b6e126ac..3edf13f7 100644 --- a/modules/mesh-grants/internal/grants/part.go +++ b/modules/mesh-grants/internal/grants/part.go @@ -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 diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index ae8842d7..36c611bb 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, + 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).