diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go index dcdd34bf..12cfabb3 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/consumes.go @@ -17,8 +17,9 @@ import ( ) // **What the watchman consumes** (its manifest's `consumes`; novox/hq ADR 0595 §4, design 25 § 12.2; DLV-W1): the indexer's -// manifest-accepted, on its durable consumer on EVENTS, and the bus module's created and refused, on its durable consumer -// on the bus module's own stream, which its membership names (Hears). Each consumer is the one the grants part makes; one +// manifest-accepted and the seat mesh-broker's created and refused, all on its durable consumer on EVENTS (the seat's +// events on the seat's subjects like every seat's: the lead's ruling, registry milestone 13:0x); beside it any module +// stream its membership names (Hears). Each consumer is the one the grants part makes; one // not there yet is opened when the watchman's membership next changes, and one opened is never opened twice. An event is // acknowledged once handed to the watchman, so one given while it was down is heard when it is back. @@ -29,8 +30,8 @@ const ( eventRefused = "refused" ) -// consumed hands one consumed event to the watchman, by its subject: manifest-accepted from EVENTS, created and refused -// from the bus module's stream. Anything else is left. +// consumed hands one consumed event to the watchman, by its subject: manifest-accepted, and the seat mesh-broker's created +// and refused. Anything else is left. func consumed(w *watch.Watchman, subject string, body []byte, logf func(string, ...any)) { switch { case strings.HasSuffix(subject, ".event."+eventAccepted) && strings.HasPrefix(subject, "mesh.mod."): @@ -110,8 +111,8 @@ func hearConsumed(ctx context.Context, js jetstream.JetStream, cred process.Acco }() } -// brokerSubject is a subject the bus module's created or refused arrives on: its own stream's, or the seat mesh-broker's -// (where the seat's definition puts it; only its holder publishes there). +// brokerSubject is a subject the bus module's created or refused arrives on: the seat mesh-broker's, where only its holder +// publishes. The bus module's own stream is its history for its tool, never heard here. func brokerSubject(subject string) bool { - return strings.HasPrefix(subject, "mesh.stream.") || strings.HasPrefix(subject, "mesh.seat."+catalogue.BrokerSeat+".event.") + return strings.HasPrefix(subject, "mesh.seat."+catalogue.BrokerSeat+".event.") } diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go index 654c757e..f4c9e742 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go @@ -52,7 +52,12 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 out := &findingsOut{findings: map[string]bool{}} w := watch.New(out, time.Now) say := func(string, ...any) {} + // The bus module's own stream is its history, never heard here: only the seat's subject (ruling 13:0x). consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l","why":"no room"}`), say) + if len(out.findings) != 0 { + t.Fatalf("a refused on the bus module's own stream reached %v", out.findings) + } + consumed(w, "mesh.seat.mesh-broker.event.refused", []byte(`{"module":"m","local":"l","why":"no room"}`), say) if !out.findings[watch.FindingKey(watch.ScopeModule, "m", watch.KindBusStateRefused+"-l")] { t.Fatalf("refused reached no finding: %v", out.findings) } @@ -60,7 +65,7 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6 if len(out.findings) != 0 { t.Fatalf("an accepted manifest no longer asking for l left %v", out.findings) } - consumed(w, "mesh.stream.nats.event.refused", []byte(`{"module":"m","local":"l"}`), say) + consumed(w, "mesh.seat.mesh-broker.event.refused", []byte(`{"module":"m","local":"l"}`), say) consumed(w, "mesh.seat.mesh-broker.event.created", []byte(`{"module":"m","local-name":"l"}`), say) if len(out.findings) != 0 { t.Fatalf("created left %v", out.findings) diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go index 8529b4c1..b387041d 100644 --- a/modules/mesh-delivery-watchman/manifest_test.go +++ b/modules/mesh-delivery-watchman/manifest_test.go @@ -103,7 +103,7 @@ func TestManifestStateAndReads(t *testing.T) { // 1135/AC6 t.Errorf("tools = %v", m.Tools) } // DLV-W1 (ADR 0595 §4): it hears the accepted verdicts and the bus module's created and refused, durably. - if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.manifest-accepted", "nats.created", "nats.refused"}) { + if !reflect.DeepEqual(m.Consumes, []string{"mesh-delivery-indexer.manifest-accepted", "mesh-broker.created", "mesh-broker.refused"}) { t.Errorf("consumes = %v", m.Consumes) } for _, k := range []string{"emits", "context", "items", "provides"} { diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json index d8d87a1a..16181760 100644 --- a/modules/mesh-delivery-watchman/module.json +++ b/modules/mesh-delivery-watchman/module.json @@ -31,8 +31,8 @@ ], "consumes": [ "mesh-delivery-indexer.manifest-accepted", - "nats.created", - "nats.refused" + "mesh-broker.created", + "mesh-broker.refused" ], "state": [ { "name": "presence", "history": 20 }, diff --git a/modules/mesh-grants/cmd/mesh-grants/integration_test.go b/modules/mesh-grants/cmd/mesh-grants/integration_test.go index b2ce0815..7f67d48d 100644 --- a/modules/mesh-grants/cmd/mesh-grants/integration_test.go +++ b/modules/mesh-grants/cmd/mesh-grants/integration_test.go @@ -9,6 +9,7 @@ import ( "io" "os" "path/filepath" + "slices" "strings" "sync/atomic" "testing" @@ -231,19 +232,16 @@ func TestTheGrantsPartWhole(t *testing.T) { if firstViewAsks.Load() < int32(len(grants.FirstViews)) { t.Errorf("the part composed with %d first-view asks, want one of each of the %d", firstViewAsks.Load(), len(grants.FirstViews)) } - // DLV-G1, 1966/AC34 (novox/hq ADR 0595 §1, §7): the bus module's created, said on its own stream, reaches the part on - // its durable consumer there, and the binding it carries is in the module's membership; dropped takes it out. - until("the part's durable consumer on the bus module's stream", func() bool { - e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a."+grants.Module) + // DLV-G1, 1966/AC34 (novox/hq ADR 0595 §1, §7): the bus module's created, said on the seat mesh-broker's subject on + // EVENTS like every seat's event (the lead's ruling, registry milestone 13:0x), reaches the part on its durable consumer + // there, and the binding it carries is in the module's membership; dropped takes it out. + until("the part's durable consumer on EVENTS hearing mesh-broker's created", func() bool { + cons, err := js.Consumer(ctx, "EVENTS", "a_"+grants.Module) if err != nil { return false } - m, err := membership.DecodeGrantsMembership(e.Value()) - if err != nil || len(m.Hears) == 0 { - return false - } - _, err = js.Consumer(ctx, m.Hears[0].Stream, m.Hears[0].Consumer) - return err == nil + info, err := cons.Info(ctx) + return err == nil && slices.Contains(info.Config.FilterSubjects, "mesh.seat.mesh-broker.event.created") }) bindingOf := func() []membership.BusBinding { e, err := kvs[grants.Module+"."+grants.StateMemberships].Get(ctx, "a.mod-c") @@ -255,7 +253,7 @@ func TestTheGrantsPartWhole(t *testing.T) { } sayBroker := func(event string, bind membership.BusBinding) { body, _ := json.Marshal(grants.BrokerEvent{V: 1, Module: "mod-c", Local: "kept", Storage: "key-value", Template: "state", Binding: bind}) - if _, err := js.Publish(ctx, "mesh.stream.nats.event."+event, body); err != nil { + if _, err := js.Publish(ctx, "mesh.seat.mesh-broker.event."+event, body); err != nil { t.Fatal(err) } } diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 584e3190..716be76e 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -2238,7 +2238,8 @@ func TestGrants(t *testing.T) { return changed } first := membership.BusBinding{Local: "records", Storage: "key-value", Template: "state", Bucket: "mod-x_records"} - for _, event := range []string{"mesh.stream.nats.event.refused", "mesh.stream.mod-y.event.created", "mesh.stream.nats.event.drifted"} { + for _, event := range []string{"mesh.seat.mesh-broker.event.refused", "mesh.stream.mod-y.event.created", + "mesh.seat.mesh-broker.event.drifted", "mesh.stream.nats.event.created", "mesh.mod.nats.event.created", "nats.created"} { if hear(event, first) { t.Errorf("%s changed the bindings", event) } @@ -2253,7 +2254,7 @@ func TestGrants(t *testing.T) { if !slices.Contains(heard.consumerCreates, "THINGS.b_mod-y") { t.Errorf("the consumer of a heard module stream was not created: %v", heard.consumerCreates) } - if !hear("mesh.stream.nats.event.created", first) || hear("mesh.stream.nats.event.created", first) { + if !hear("mesh.seat.mesh-broker.event.created", first) || hear("mesh.seat.mesh-broker.event.created", first) { t.Fatal("created: not kept once") } b.assign("b", "mod-x") @@ -2268,7 +2269,7 @@ func TestGrants(t *testing.T) { } again := first again.Bucket = "mod-x_records-2" - hear("mesh.stream.nats.event.recreated", again) + hear("mesh.seat.mesh-broker.event.recreated", again) p.Compose() if got := bindingsOn("b.mod-x"); len(got) != 1 || got[0].Bucket != "mod-x_records-2" { t.Errorf("after recreated: %+v", got) diff --git a/modules/mesh-grants/internal/buscompose/buscompose_test.go b/modules/mesh-grants/internal/buscompose/buscompose_test.go index 59c6bbc6..304636a2 100644 --- a/modules/mesh-grants/internal/buscompose/buscompose_test.go +++ b/modules/mesh-grants/internal/buscompose/buscompose_test.go @@ -1069,12 +1069,14 @@ func aLedgerIsWrittenByItsLedgerAccountAlone(t *testing.T) { func theBusModulesProcessKeepsBusState(t *testing.T) { r := Records{Nodes: []string{"a"}, Assigned: map[string][]Declared{"a": { {Module: "node-tools", ToolRunner: true}, - {Module: "nats", SnapshotsTheBus: true, Holds: []Seat{{Name: catalogue.BrokerSeat, Serves: BrokerVerbs()}}, + {Module: "nats", SnapshotsTheBus: true, Holds: []Seat{{Name: catalogue.BrokerSeat, Serves: BrokerVerbs(), + Emits: []string{"created", "refused"}}}, Consumes: []string{"mesh-delivery-indexer.manifest-accepted"}, Uses: []Seat{{Name: "operator-channel", Accepts: []string{"condition"}, ByCaller: []string{"condition"}}}, Streams: []ModuleStream{{Module: "nats", Name: "BROKER_EVENTS", Events: []string{"created", "refused"}, MaxAgeDays: 30, MaxMiB: 256}}, Processes: []ProcessPrincipal{{Process: "keeper", State: []string{"asked"}, Keeps: true}, {Process: "other"}, {Process: "nats-tools", Tools: true, Removes: []string{catalogue.RemovesEngineActs}}}}, - {Module: "mod-x", Processes: []ProcessPrincipal{{Process: "sneak", Keeps: true}}}, + {Module: "mod-x", Processes: []ProcessPrincipal{{Process: "sneak", Keeps: true}}, + Streams: []ModuleStream{{Module: "mod-x", Name: "SECRETS", Events: []string{"sealed"}, Restricted: true}}}, {Module: "admin-ish", Invokes: []string{"*"}}, }}, BusState: map[string][]catalogue.BusStateEntry{"mod-x": { @@ -1132,11 +1134,39 @@ func theBusModulesProcessKeepsBusState(t *testing.T) { t.Error("the bus module's process answers what it was asked") } for _, s := range []string{"$JS.API.CONSUMER.MSG.NEXT.EVENTS.a_nats_keeper", "$JS.ACK.EVENTS.a_nats_keeper.x", - "mesh.stream.nats.event.created", "mesh.seat.operator-channel.accept.condition.nats"} { + "mesh.stream.nats.event.created", "mesh.seat.operator-channel.accept.condition.nats", + // The seat's events on the seat's subjects, as every seat's (the lead's ruling, registry milestone 13:0x), and the + // acknowledgement of a snapshot's chunks of an entry it keeps (ADR 0626 §1). + "mesh.seat.mesh-broker.event.created", "mesh.seat.mesh-broker.event.refused", "$JS.SNAPSHOT.ACK.KV_mod-x_records.x.1.0"} { if !MayPublish(keeper, s) { t.Errorf("the bus module's process may not publish %s", s) } } + for _, other := range []string{"a.nats.other", "a.mod-x.sneak"} { + if MayPublish(perms[other], "mesh.seat.mesh-broker.event.created") { + t.Errorf("%s, which is no keeper of the bus, says mesh-broker's created", other) + } + } + // ADR 0626, How it is checked: no subscription to a restricted stream's subjects, nor a consumer of it or of any + // stream it keeps; a snapshot's chunks come to its own inbox alone, and only of a stream it keeps. + for _, s := range []string{"mesh.stream.mod-x.event.sealed", "mesh.stream.mod-x.event.>", "mesh.bus.mod-x.records.x"} { + if MaySubscribe(keeper, s) { + t.Errorf("the bus module's process hears %s", s) + } + } + for _, s := range []string{"$JS.API.CONSUMER.CREATE.SECRETS.x.y", "$JS.API.CONSUMER.CREATE.KV_mod-x_records.x", + "$JS.API.CONSUMER.MSG.NEXT.SECRETS.x", "$JS.API.DIRECT.GET.SECRETS", "$JS.API.STREAM.SNAPSHOT.SECRETS", + "$JS.SNAPSHOT.ACK.SECRETS.x.1.0", "$JS.API.STREAM.MSG.GET.KV_mod-x_records"} { + if MayPublish(keeper, s) { + t.Errorf("the bus module's process may publish %s", s) + } + } + for _, s := range keeper.Subscribe { + if !strings.HasPrefix(s, "_INBOX.a.nats.keeper.") && !strings.HasPrefix(s, "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_nats_") && + !strings.HasPrefix(s, "mesh.seat.operator-channel.") { + t.Errorf("the bus module's process subscribes %s, beyond its inbox, its own state's creation and its conditions' answers", s) + } + } made := false for _, mc := range AccountConsumers(users) { if mc.Consumer.Stream == "EVENTS" && mc.Consumer.Name == "a_nats_keeper" && diff --git a/modules/mesh-grants/internal/buscompose/busstate.go b/modules/mesh-grants/internal/buscompose/busstate.go index 0f9197ec..27b73d5b 100644 --- a/modules/mesh-grants/internal/buscompose/busstate.go +++ b/modules/mesh-grants/internal/buscompose/busstate.go @@ -144,13 +144,34 @@ func BusStateNames(asked map[string][]catalogue.BusStateEntry, bound map[string] } // busStateKeeperGrants are what the bus module's process publishes to keep the bus state named: each stream's create, -// update, information, deletion and snapshot, and the stream list it finds orphans in. +// update, information, deletion and snapshot with the acknowledgement of the snapshot's chunks (ADR 0626 §1; the server +// waits on `$JS.SNAPSHOT.ACK..…` past its window), and the stream list it finds orphans in. Never a consumer of +// one, nor any stream's own subjects: a snapshot carries an entry's data to the keeper's inbox alone, and the keeper +// hears no message of it (ADR 0626 §1, restricted streams included). func busStateKeeperGrants(names []string) []string { out := []string{"$JS.API.STREAM.NAMES", "$JS.API.STREAM.LIST"} for _, n := range names { for _, op := range []string{"CREATE", "UPDATE", "INFO", "DELETE", "SNAPSHOT"} { out = append(out, "$JS.API.STREAM."+op+"."+n) } + out = append(out, "$JS.SNAPSHOT.ACK."+n+".>") + } + return out +} + +// brokerSeatEvents are the subjects of the seat mesh-broker's events, from the held seat's definition: what the bus +// module's keeper says them on, like every seat's holder (the lead's ruling, registry milestone 13:0x). +func brokerSeatEvents(holds []Seat) []string { + var out []string + for _, s := range holds { + if s.Name != catalogue.BrokerSeat { + continue + } + for _, e := range s.Emits { + if safeSubject.MatchString(e) { + out = append(out, seatSubject(s, "event", e)) + } + } } return out } diff --git a/modules/mesh-grants/internal/buscompose/nats.go b/modules/mesh-grants/internal/buscompose/nats.go index b56873b0..e029a421 100644 --- a/modules/mesh-grants/internal/buscompose/nats.go +++ b/modules/mesh-grants/internal/buscompose/nats.go @@ -114,6 +114,10 @@ type Principal struct { // §4, §5), and KeepsBusState are the streams it defines, every module's bus state by its name. KeepsBus bool KeepsBusState []string + // SaysSeatEvents are the subjects of the seat mesh-broker's events the bus module's keeper says them on (the lead's + // ruling, registry milestone 13:0x): `mesh.seat.mesh-broker.event.` for each event the seat's definition + // names, as a holder's own account would be granted them; set on the keeper alone. + SaysSeatEvents []string // Process is the process resource's id, for a KindProcess principal alone (novox/hq ADR 0436 §2): // Username and inbox are ... Process string diff --git a/modules/mesh-grants/internal/buscompose/process_grants.go b/modules/mesh-grants/internal/buscompose/process_grants.go index 26f81c37..e6509a3c 100644 --- a/modules/mesh-grants/internal/buscompose/process_grants.go +++ b/modules/mesh-grants/internal/buscompose/process_grants.go @@ -61,6 +61,7 @@ func processGrants(p Principal) (pub, sub []string, err error) { pub = append(pub, keeperConsumerGrants(c)...) } pub = append(pub, ownStreamGrants(p.Module, p.Streams)...) + pub = append(pub, p.SaysSeatEvents...) tp, ts := seatTrafficGrants(SeatTrafficOf(p.Module, nil, p.Uses, nil)) pub, sub = append(pub, tp...), append(sub, ts...) } diff --git a/modules/mesh-grants/internal/buscompose/users.go b/modules/mesh-grants/internal/buscompose/users.go index 6dceb5d7..5d521da9 100644 --- a/modules/mesh-grants/internal/buscompose/users.go +++ b/modules/mesh-grants/internal/buscompose/users.go @@ -249,6 +249,7 @@ func Users(r Records) ([]Principal, error) { k := &out[len(out)-1] k.KeepsBus, k.KeepsBusState = true, BusStateNames(r.BusState, r.Bound) k.Consumes, k.Streams = d.Consumes, d.Streams + k.SaysSeatEvents = brokerSeatEvents(d.Holds) } if d.SnapshotsTheBus && len(pr.Removes) > 0 { out[len(out)-1].Streams = d.Streams diff --git a/modules/mesh-grants/internal/grants/bindings.go b/modules/mesh-grants/internal/grants/bindings.go index badc7e04..a06c7922 100644 --- a/modules/mesh-grants/internal/grants/bindings.go +++ b/modules/mesh-grants/internal/grants/bindings.go @@ -61,28 +61,13 @@ type Bound struct { Bindings map[string]membership.BusBinding `json:"bindings"` } -// brokerEvent is the local name of an event heard from the holder of mesh-broker, by its name or its bus subject, and -// whether it is one; an emitter that holds no mesh-broker seat is never one. +// brokerEvent is the local name of an event heard from the holder of mesh-broker, by its bus subject, and whether it is +// one: the seat's own subject alone (`mesh.seat.mesh-broker.event.`, the lead's ruling, registry milestone 13:0x), +// where only its holder publishes, and only while a module holds the seat. The bus module's own stream is its history +// for its tool, never what the part hears. func brokerEvent(event string, brokers map[string]bool) (string, bool) { - emitter, name := "", "" - switch { - case strings.HasPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event."): - // The seat's own subject: only its holder publishes there, so the emitter is the holder. - name = strings.TrimPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event.") - for b := range brokers { - emitter = b - } - case buscompose.OnAModuleStream(event): - // mesh.stream..event. - rest := strings.TrimPrefix(event, "mesh.stream.") - emitter, name, _ = strings.Cut(rest, ".event.") - case strings.HasPrefix(event, "mesh.mod."): - rest := strings.TrimPrefix(event, "mesh.mod.") - emitter, name, _ = strings.Cut(rest, ".event.") - default: - emitter, name, _ = strings.Cut(event, ".") - } - if !brokers[emitter] { + name, onTheSeat := strings.CutPrefix(event, "mesh.seat."+catalogue.BrokerSeat+".event.") + if !onTheSeat || len(brokers) == 0 { return "", false } switch name { diff --git a/modules/mesh-grants/internal/grants/rootfree.go b/modules/mesh-grants/internal/grants/rootfree.go index da933b8d..e76fbbaa 100644 --- a/modules/mesh-grants/internal/grants/rootfree.go +++ b/modules/mesh-grants/internal/grants/rootfree.go @@ -7,6 +7,7 @@ import ( "sort" "strings" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/health" "mesh-grants/internal/buscompose" ) @@ -39,8 +40,10 @@ func FromKeeper(event string) bool { const EventAccountIssued = "mesh-vault." + buscompose.AccountIssuedEvent // Consumes is every event the part hears, as its manifest's consumes name them. -var Consumes = append([]string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, BusModule + "." + BrokerCreated, - BusModule + "." + BrokerDropped, BusModule + "." + BrokerRecreated}, viewsEvents()...) +// The seat mesh-broker's created, recreated and dropped are heard as every seat's event, `.` on EVENTS (the +// lead's ruling, registry milestone 13:0x), never on the bus module's own stream. +var Consumes = append([]string{EventKeeperRaised, EventKeeperCleared, EventAccountIssued, catalogue.BrokerSeat + "." + BrokerCreated, + catalogue.BrokerSeat + "." + BrokerDropped, catalogue.BrokerSeat + "." + BrokerRecreated}, viewsEvents()...) // viewsEvents are the owners' events the part follows its first views by: `.views.`, but the verdicts by // the indexer's manifest-accepted and manifest-refused (VerdictEvents, ADR 0622 §2). diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 3819e9aa..a982d3c6 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -14,9 +14,9 @@ "mesh-conditions.condition-raised", "mesh-conditions.condition-cleared", "mesh-vault.account-issued", - "nats.created", - "nats.dropped", - "nats.recreated", + "mesh-broker.created", + "mesh-broker.dropped", + "mesh-broker.recreated", "mesh-delivery-assigner.views.placement", "mesh-delivery-assigner.views.wants", "mesh-delivery-indexer.views.manifests",